X-Git-Url: http://git.indexdata.com/?a=blobdiff_plain;f=src%2Flogic.c;h=2ee120341b2f034426f7d7b97aaad8d72e5794b1;hb=5c017d34db6ea43ee6a95b8c462b9b824c26fbb8;hp=a882ac34154b4eb8adc9ba93910b9cddeb0a3f7e;hpb=b69ffeb3020d001f75730fe48fa9e462761c69e1;p=pazpar2-moved-to-github.git

diff --git a/src/logic.c b/src/logic.c
index a882ac3..2ee1203 100644
--- a/src/logic.c
+++ b/src/logic.c
@@ -1,5 +1,5 @@
 /* This file is part of Pazpar2.
-   Copyright (C) 2006-2008 Index Data
+   Copyright (C) 2006-2009 Index Data
 
 Pazpar2 is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free
@@ -57,7 +57,7 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 #include <yaz/timing.h>
 #endif
 
-
+#include "parameters.h"
 #include "pazpar2.h"
 #include "eventl.h"
 #include "http.h"
@@ -76,25 +76,28 @@ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 // Note: Some things in this structure will eventually move to configuration
 struct parameters global_parameters = 
 {
-    "",
-    "",
-    "", 
-    0,
     0,   // dump_records
-    0,   // debug_mode
-    30,  // operations timeout 
-    "81",
-    "Index Data PazPar2",
-    VERSION,
-    60,   // session timeout 
-    100,
-    MAX_CHUNK,
-    0,
-    0,
-    180, // Z39.50 session timeout
-    15   // Connect timeout
+    0    // debug_mode
 };
 
+static void log_xml_doc(xmlDoc *doc)
+{
+    FILE *lf = yaz_log_file();
+    xmlChar *result = 0;
+    int len = 0;
+#if LIBXML_VERSION >= 20600
+    xmlDocDumpFormatMemory(doc, &result, &len, 1);
+#else
+    xmlDocDumpMemory(doc, &result, &len);
+#endif
+    if (lf && len)
+    {
+        fwrite(result, 1, len, lf);
+        fprintf(lf, "\n");
+    }
+    xmlFree(result);
+}
+
 // Recursively traverse query structure to extract terms.
 void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 {
@@ -142,14 +145,13 @@ static void add_facet(struct session *s, const char *type, const char *value)
 
         s->termlists[i].name = nmem_strdup(s->nmem, type);
         s->termlists[i].termlist 
-            = termlist_create(s->nmem, s->expected_maxrecs,
-                              TERMLIST_HIGH_SCORE);
+            = termlist_create(s->nmem, TERMLIST_HIGH_SCORE);
         s->num_termlists = i + 1;
     }
     termlist_insert(s->termlists[i].termlist, value);
 }
 
-xmlDoc *record_to_xml(struct session_database *sdb, const char *rec)
+static xmlDoc *record_to_xml(struct session_database *sdb, const char *rec)
 {
     struct database *db = sdb->database;
     xmlDoc *rdoc = 0;
@@ -165,17 +167,8 @@ xmlDoc *record_to_xml(struct session_database *sdb, const char *rec)
 
     if (global_parameters.dump_records)
     {
-        FILE *lf = yaz_log_file();
-        if (lf)
-        {
-            yaz_log(YLOG_LOG, "Un-normalized record from %s", db->url);
-#if LIBXML_VERSION >= 20600
-            xmlDocFormatDump(lf, rdoc, 1);
-#else
-            xmlDocDump(lf, rdoc);
-#endif
-            fprintf(lf, "\n");
-        }
+        yaz_log(YLOG_LOG, "Un-normalized record from %s", db->url);
+        log_xml_doc(rdoc);
     }
 
     return rdoc;
@@ -187,7 +180,7 @@ xmlDoc *record_to_xml(struct session_database *sdb, const char *rec)
 static void insert_settings_parameters(struct session_database *sdb,
                                        struct session *se, char **parms)
 {
-    struct conf_service *service = global_parameters.server->service;
+    struct conf_service *service = se->service;
     int i;
     int nparms = 0;
     int offset = 0;
@@ -198,7 +191,7 @@ static void insert_settings_parameters(struct session_database *sdb,
         int setting;
 
         if (md->setting == Metadata_setting_parameter &&
-            (setting = settings_offset(md->name)) > 0)
+            (setting = settings_offset(service, md->name)) > 0)
         {
             const char *val = session_setting_oneval(sdb, setting);
             if (val && nparms < MAX_XSLT_ARGS)
@@ -220,9 +213,9 @@ static void insert_settings_parameters(struct session_database *sdb,
 }
 
 // Add static values from session database settings if applicable
-static void insert_settings_values(struct session_database *sdb, xmlDoc *doc)
+static void insert_settings_values(struct session_database *sdb, xmlDoc *doc,
+    struct conf_service *service)
 {
-    struct conf_service *service = global_parameters.server->service;
     int i;
 
     for (i = 0; i < service->num_metadata; i++)
@@ -231,7 +224,7 @@ static void insert_settings_values(struct session_database *sdb, xmlDoc *doc)
         int offset;
 
         if (md->setting == Metadata_setting_postproc &&
-            (offset = settings_offset(md->name)) > 0)
+            (offset = settings_offset(service, md->name)) > 0)
         {
             const char *val = session_setting_oneval(sdb, offset);
             if (val)
@@ -245,55 +238,31 @@ static void insert_settings_values(struct session_database *sdb, xmlDoc *doc)
     }
 }
 
-xmlDoc *normalize_record(struct session_database *sdb, struct session *se,
-                         const char *rec)
+static xmlDoc *normalize_record(struct session_database *sdb,
+                                struct session *se,
+                                const char *rec)
 {
-    struct database_retrievalmap *m;
     xmlDoc *rdoc = record_to_xml(sdb, rec);
+
     if (rdoc)
     {
-        for (m = sdb->map; m; m = m->next)
+        char *parms[MAX_XSLT_ARGS*2+1];
+        
+        insert_settings_parameters(sdb, se, parms);
+        
+        if (normalize_record_transform(sdb->map, &rdoc, (const char **)parms))
         {
-            xmlDoc *new = 0;
-            
-            {
-                xmlNodePtr root = 0;
-                char *parms[MAX_XSLT_ARGS*2+1];
-
-                insert_settings_parameters(sdb, se, parms);
-
-                new = xsltApplyStylesheet(m->stylesheet, rdoc, (const char **) parms);
-                root= xmlDocGetRootElement(new);
-                if (!new || !root || !(root->children))
-                {
-                    yaz_log(YLOG_WARN, "XSLT transformation failed from %s",
-                            sdb->database->url);
-                    xmlFreeDoc(new);
-                    xmlFreeDoc(rdoc);
-                    return 0;
-                }
-            }
-            
-            xmlFreeDoc(rdoc);
-            rdoc = new;
+            yaz_log(YLOG_WARN, "Normalize failed from %s", sdb->database->url);
         }
-
-        insert_settings_values(sdb, rdoc);
-
-        if (global_parameters.dump_records)
+        else
         {
-            FILE *lf = yaz_log_file();
+            insert_settings_values(sdb, rdoc, se->service);
             
-            if (lf)
+            if (global_parameters.dump_records)
             {
                 yaz_log(YLOG_LOG, "Normalized record from %s", 
                         sdb->database->url);
-#if LIBXML_VERSION >= 20600
-                xmlDocFormatDump(lf, rdoc, 1);
-#else
-                xmlDocDump(lf, rdoc);
-#endif
-                fprintf(lf, "\n");
+                log_xml_doc(rdoc);
             }
         }
     }
@@ -325,9 +294,6 @@ static int prepare_map(struct session *se, struct session_database *sdb)
     }
     if ((s = session_setting_oneval(sdb, PZ_XSLT)))
     {
-        char **stylesheets;
-        struct database_retrievalmap **m = &sdb->map;
-        int num, i;
         char auto_stylesheet[256];
 
         if (!strcmp(s, "auto"))
@@ -352,23 +318,11 @@ static int prepare_map(struct session *se, struct session_database *sdb)
                 yaz_log(YLOG_WARN, "No pz:requestsyntax for auto stylesheet");
             }
         }
-        nmem_strsplit(se->session_nmem, ",", s, &stylesheets, &num);
-        for (i = 0; i < num; i++)
-        {
-            (*m) = nmem_malloc(se->session_nmem, sizeof(**m));
-            (*m)->next = 0;
-            if (!((*m)->stylesheet = conf_load_stylesheet(stylesheets[i])))
-            {
-                yaz_log(YLOG_FATAL|YLOG_ERRNO, "Unable to load stylesheet: %s",
-                        stylesheets[i]);
-                return -1;
-            }
-            m = &(*m)->next;
-        }
+        sdb->map = normalize_cache_get(se->normalize_cache,
+                                       se->service, s);
+        if (!sdb->map)
+            return -1;
     }
-    if (!sdb->map)
-        yaz_log(YLOG_WARN, "No Normalization stylesheet for target %s",
-                sdb->database->url);
     return 0;
 }
 
@@ -487,10 +441,14 @@ static struct database_criterion *parse_filter(NMEM m, const char *buf)
         int subnum;
         int subi;
         struct database_criterion *new = nmem_malloc(m, sizeof(*new));
-        char *eq = strchr(values[i], '=');
-        if (!eq)
+        char *eq;
+        if ((eq = strchr(values[i], '=')))
+            new->type = PAZPAR2_STRING_MATCH;
+        else if ((eq = strchr(values[i], '~')))
+            new->type = PAZPAR2_SUBSTRING_MATCH;
+        else
         {
-            yaz_log(YLOG_WARN, "Missing equal-sign in filter");
+            yaz_log(YLOG_WARN, "Missing equal-sign/tilde in filter");
             return 0;
         }
         *(eq++) = '\0';
@@ -512,7 +470,9 @@ static struct database_criterion *parse_filter(NMEM m, const char *buf)
 }
 
 enum pazpar2_error_code search(struct session *se,
-                               char *query, char *filter,
+                               const char *query,
+                               const char *maxrecs,
+                               const char *filter,
                                const char **addinfo)
 {
     int live_channels = 0;
@@ -533,36 +493,37 @@ enum pazpar2_error_code search(struct session *se,
     live_channels = select_targets(se, criteria);
     if (live_channels)
     {
-        int maxrecs = live_channels * global_parameters.toget;
-        se->reclist = reclist_create(se->nmem, maxrecs);
-        se->expected_maxrecs = maxrecs;
+        se->reclist = reclist_create(se->nmem);
     }
     else
         return PAZPAR2_NO_TARGETS;
 
     for (cl = se->clients; cl; cl = client_next_in_session(cl))
     {
+        if (maxrecs)
+            client_set_maxrecs(cl, atoi(maxrecs));
         if (prepare_session_database(se, client_get_database(cl)) < 0)
-        {
-            *addinfo = client_get_database(cl)->database->url;
-            return PAZPAR2_CONFIG_TARGET;
-        }
+            continue;
         // Parse query for target
         if (client_parse_query(cl, query) < 0)
             no_failed++;
         else
         {
             no_working++;
-            if (client_prep_connection(cl))
+            if (client_prep_connection(cl, se->service->z3950_operation_timeout,
+                                       se->service->z3950_session_timeout))
                 client_start_search(cl);
         }
     }
-
-    // If no queries could be mapped, we signal an error
     if (no_working == 0)
     {
-        *addinfo = "query";
-        return PAZPAR2_MALFORMED_PARAMETER_VALUE;
+        if (no_failed > 0)
+        {
+            *addinfo = "query";
+            return PAZPAR2_MALFORMED_PARAMETER_VALUE;
+        }
+        else
+            return PAZPAR2_NO_TARGETS;
     }
     return PAZPAR2_NO_ERROR;
 }
@@ -571,8 +532,9 @@ enum pazpar2_error_code search(struct session *se,
 static void session_init_databases_fun(void *context, struct database *db)
 {
     struct session *se = (struct session *) context;
+    struct conf_service *service = se->service;
     struct session_database *new = nmem_malloc(se->session_nmem, sizeof(*new));
-    int num = settings_num();
+    int num = settings_num(service);
     int i;
 
     new->database = db;
@@ -594,10 +556,7 @@ static void session_init_databases_fun(void *context, struct database *db)
 // Doesn't free memory associated with sdb -- nmem takes care of that
 static void session_database_destroy(struct session_database *sdb)
 {
-    struct database_retrievalmap *m;
-
-    for (m = sdb->map; m; m = m->next)
-        xsltFreeStylesheet(m->stylesheet);
+    sdb->map = 0;
 }
 
 // Initialize session_database list -- this represents this session's view
@@ -605,7 +564,7 @@ static void session_database_destroy(struct session_database *sdb)
 void session_init_databases(struct session *se)
 {
     se->databases = 0;
-    predef_grep_databases(se, 0, session_init_databases_fun);
+    predef_grep_databases(se, se->service, 0, session_init_databases_fun);
 }
 
 // Probably session_init_databases_fun should be refactored instead of
@@ -613,7 +572,9 @@ void session_init_databases(struct session *se)
 static struct session_database *load_session_database(struct session *se, 
                                                       char *id)
 {
-    struct database *db = find_database(id, 0);
+    struct database *db = find_database(id, 0, se->service);
+
+    resolve_database(db);
 
     session_init_databases_fun((void*) se, db);
     // New sdb is head of se->databases list
@@ -637,8 +598,9 @@ void session_apply_setting(struct session *se, char *dbname, char *setting,
                            char *value)
 {
     struct session_database *sdb = find_session_database(se, dbname);
+    struct conf_service *service = se->service;
     struct setting *new = nmem_malloc(se->session_nmem, sizeof(*new));
-    int offset = settings_offset_cprefix(setting);
+    int offset = settings_offset_cprefix(service, setting);
 
     if (offset < 0)
     {
@@ -665,10 +627,6 @@ void session_apply_setting(struct session *se, char *dbname, char *setting,
     case PZ_XSLT:
         if (sdb->map)
         {
-            struct database_retrievalmap *m;
-            // We don't worry about the map structure -- it's in nmem
-            for (m = sdb->map; m; m = m->next)
-                xsltFreeStylesheet(m->stylesheet);
             sdb->map = 0;
         }
         break;
@@ -683,25 +641,28 @@ void destroy_session(struct session *s)
         client_destroy(s->clients);
     for (sdb = s->databases; sdb; sdb = sdb->next)
         session_database_destroy(sdb);
+    normalize_cache_destroy(s->normalize_cache);
     nmem_destroy(s->nmem);
+    service_destroy(s->service);
     wrbuf_destroy(s->wrbuf);
 }
 
-struct session *new_session(NMEM nmem) 
+struct session *new_session(NMEM nmem, struct conf_service *service) 
 {
     int i;
     struct session *session = nmem_malloc(nmem, sizeof(*session));
 
     yaz_log(YLOG_DEBUG, "New Pazpar2 session");
-    
+
+    session->service = service;
     session->relevance = 0;
     session->total_hits = 0;
     session->total_records = 0;
     session->number_of_warnings_unknown_elements = 0;
+    session->number_of_warnings_unknown_metadata = 0;
     session->num_termlists = 0;
     session->reclist = 0;
     session->clients = 0;
-    session->expected_maxrecs = 0;
     session->session_nmem = nmem;
     session->nmem = nmem_create();
     session->wrbuf = wrbuf_alloc();
@@ -711,6 +672,8 @@ struct session *new_session(NMEM nmem)
         session->watchlist[i].data = 0;
         session->watchlist[i].fun = 0;
     }
+    session->normalize_cache = normalize_cache_create();
+
     return session;
 }
 
@@ -765,14 +728,24 @@ void report_nmem_stats(void)
 }
 #endif
 
-struct record_cluster *show_single(struct session *s, const char *id)
+struct record_cluster *show_single(struct session *s, const char *id,
+                                   struct record_cluster **prev_r,
+                                   struct record_cluster **next_r)
 {
     struct record_cluster *r;
 
     reclist_rewind(s->reclist);
+    *prev_r = 0;
+    *next_r = 0;
     while ((r = reclist_read_record(s->reclist)))
+    {
         if (!strcmp(r->recid, id))
+        {
+            *next_r = reclist_read_record(s->reclist);
             return r;
+        }
+        *prev_r = r;
+    }
     return 0;
 }
 
@@ -805,7 +778,7 @@ struct record_cluster **show(struct session *s, struct reclist_sortparms *sp,
             }
         reclist_sort(s->reclist, sp);
         
-        *total = s->reclist->num_records;
+        *total = reclist_get_num_records(s->reclist);
         *sumhits = s->total_hits;
         
         for (i = 0; i < start; i++)
@@ -864,52 +837,10 @@ void statistics(struct session *se, struct statistics *stat)
     stat->num_clients = count;
 }
 
-int start_http_listener(void)
-{
-    char hp[128] = "";
-    struct conf_server *ser = global_parameters.server;
-
-    if (*global_parameters.listener_override)
-        strcpy(hp, global_parameters.listener_override);
-    else
-    {
-        strcpy(hp, ser->host ? ser->host : "");
-        if (ser->port)
-        {
-            if (*hp)
-                strcat(hp, ":");
-            sprintf(hp + strlen(hp), "%d", ser->port);
-        }
-    }
-    return http_init(hp);
-}
-
-void start_proxy(void)
-{
-    char hp[128] = "";
-    struct conf_server *ser = global_parameters.server;
-
-    if (*global_parameters.proxy_override)
-        strcpy(hp, global_parameters.proxy_override);
-    else if (ser->proxy_host || ser->proxy_port)
-    {
-        strcpy(hp, ser->proxy_host ? ser->proxy_host : "");
-        if (ser->proxy_port)
-        {
-            if (*hp)
-                strcat(hp, ":");
-            sprintf(hp + strlen(hp), "%d", ser->proxy_port);
-        }
-    }
-    else
-        return;
-
-    http_set_proxyaddr(hp, ser->myurl ? ser->myurl : "");
-}
-
 
 // Master list of connections we're handling events to
-static IOCHAN channel_list = 0; 
+static IOCHAN channel_list = 0;  /* thread pr */
+
 void pazpar2_add_channel(IOCHAN chan)
 {
     chan->next = channel_list;
@@ -951,202 +882,352 @@ static struct record_metadata *record_metadata_init(
     return rec_md;
 }
 
+static int get_mergekey_from_doc(xmlDoc *doc, xmlNode *root, const char *name,
+                                 struct conf_service *service, WRBUF norm_wr)
+{
+    xmlNode *n;
+    int no_found = 0;
+    for (n = root->children; n; n = n->next)
+    {
+        if (n->type != XML_ELEMENT_NODE)
+            continue;
+        if (!strcmp((const char *) n->name, "metadata"))
+        {
+            xmlChar *type = xmlGetProp(n, (xmlChar *) "type");
+            if (!strcmp(name, (const char *) type))
+            {
+                xmlChar *value = xmlNodeListGetString(doc, n->children, 1);
+                if (value)
+                {
+                    const char *norm_str;
+                    pp2_relevance_token_t prt =
+                        pp2_relevance_tokenize(
+                            service->mergekey_pct,
+                            (const char *) value);
+                    
+                    wrbuf_puts(norm_wr, name);
+                    wrbuf_puts(norm_wr, "=");
+                    while ((norm_str =
+                            pp2_relevance_token_next(prt)))
+                    {
+                        if (*norm_str)
+                        {
+                            if (wrbuf_len(norm_wr))
+                                wrbuf_puts(norm_wr, " ");
+                            wrbuf_puts(norm_wr, norm_str);
+                        }
+                    }
+                    xmlFree(value);
+                    pp2_relevance_token_destroy(prt);
+                    no_found++;
+                }
+            }
+            xmlFree(type);
+        }
+    }
+    return no_found;
+}
+
+static const char *get_mergekey(xmlDoc *doc, struct client *cl, int record_no,
+                                struct conf_service *service, NMEM nmem)
+{
+    char *mergekey_norm = 0;
+    xmlNode *root = xmlDocGetRootElement(doc);
+    WRBUF norm_wr = wrbuf_alloc();
+
+    /* consider mergekey from XSL first */
+    xmlChar *mergekey = xmlGetProp(root, (xmlChar *) "mergekey");
+    if (mergekey)
+    {
+        const char *norm_str;
+        pp2_relevance_token_t prt =
+            pp2_relevance_tokenize(
+                service->mergekey_pct,
+                (const char *) mergekey);
+        
+        while ((norm_str = pp2_relevance_token_next(prt)))
+        {
+            if (*norm_str)
+            {
+                if (wrbuf_len(norm_wr))
+                    wrbuf_puts(norm_wr, " ");
+                wrbuf_puts(norm_wr, norm_str);
+            }
+        }
+        pp2_relevance_token_destroy(prt);
+        xmlFree(mergekey);
+    }
+    else
+    {
+        /* no mergekey defined in XSL. Look for mergekey metadata instead */
+        int field_id;
+        for (field_id = 0; field_id < service->num_metadata; field_id++)
+        {
+            struct conf_metadata *ser_md = &service->metadata[field_id];
+            if (ser_md->mergekey != Metadata_mergekey_no)
+            {
+                int r = get_mergekey_from_doc(doc, root, ser_md->name,
+                                              service, norm_wr);
+                if (r == 0 && ser_md->mergekey == Metadata_mergekey_required)
+                {
+                    /* no mergekey on this one and it is required.. 
+                       Generate unique key instead */
+                    wrbuf_rewind(norm_wr);
+                    break;
+                }
+            }
+        }
+    }
+
+    /* generate unique key if none is not generated already or is empty */
+    if (wrbuf_len(norm_wr) == 0)
+    {
+        wrbuf_printf(norm_wr, "%s-%d",
+                     client_get_database(cl)->database->url, record_no);
+    }
+    if (wrbuf_len(norm_wr) > 0)
+        mergekey_norm = nmem_strdup(nmem, wrbuf_cstr(norm_wr));
+    wrbuf_destroy(norm_wr);
+    return mergekey_norm;
+}
+
+/** \brief see if metadata for pz:recordfilter exists 
+    \param root xml root element of normalized record
+    \param sdb session database for client
+    \retval 0 if there is no metadata for pz:recordfilter
+    \retval 1 if there is metadata for pz:recordfilter
+
+    If there is no pz:recordfilter defined, this function returns 1
+    as well.
+*/
+    
+static int check_record_filter(xmlNode *root, struct session_database *sdb)
+{
+    int match = 0;
+    xmlNode *n;
+    const char *s;
+    s = session_setting_oneval(sdb, PZ_RECORDFILTER);
+
+    if (!s || !*s)
+        return 1;
+
+    for (n = root->children; n; n = n->next)
+    {
+        if (n->type != XML_ELEMENT_NODE)
+            continue;
+        if (!strcmp((const char *) n->name, "metadata"))
+        {
+            xmlChar *type = xmlGetProp(n, (xmlChar *) "type");
+            if (type)
+            {
+                size_t len;
+                const char *eq = strchr(s, '=');
+                if (eq)
+                    len = eq - s;
+                else
+                    len = strlen(s);
+                if (len == strlen((const char *)type) &&
+                    !memcmp((const char *) type, s, len))
+                {
+                    xmlChar *value = xmlNodeGetContent(n);
+                    if (value && *value)
+                    {
+                        if (!eq || strstr((const char *) value, eq+1))
+                            match = 1;
+                    }
+                    xmlFree(value);
+                }
+                xmlFree(type);
+            }
+        }
+    }
+    return match;
+}
+
+
+/** \brief ingest XML record
+    \param cl client holds the result set for record
+    \param rec record buffer (0 terminated)
+    \param record_no record position (1, 2, ..)
+    \returns resulting record or NULL on failure
+*/
 struct record *ingest_record(struct client *cl, const char *rec,
                              int record_no)
 {
-    xmlDoc *xdoc = normalize_record(client_get_database(cl),
-                                    client_get_session(cl), rec);
+    struct session_database *sdb = client_get_database(cl);
+    struct session *se = client_get_session(cl);
+    xmlDoc *xdoc = normalize_record(sdb, se, rec);
     xmlNode *root, *n;
     struct record *record;
     struct record_cluster *cluster;
-    struct session *se = client_get_session(cl);
-    xmlChar *mergekey, *mergekey_norm;
+    const char *mergekey_norm;
     xmlChar *type = 0;
     xmlChar *value = 0;
-    struct conf_service *service = global_parameters.server->service;
-    const char *norm_str = 0;
-    pp2_relevance_token_t prt = 0;
-    WRBUF norm_wr = 0;
+    struct conf_service *service = se->service;
 
     if (!xdoc)
         return 0;
 
     root = xmlDocGetRootElement(xdoc);
-    if (!(mergekey = xmlGetProp(root, (xmlChar *) "mergekey")))
+
+    if (!check_record_filter(root, sdb))
     {
-        yaz_log(YLOG_WARN, "No mergekey found in record");
+        yaz_log(YLOG_WARN, "Filtered out record no %d from %s", record_no,
+            sdb->database->url);
         xmlFreeDoc(xdoc);
         return 0;
     }
-    
-    record = record_create(se->nmem, 
-                           service->num_metadata, service->num_sortkeys, cl,
-                           record_no);
 
-    prt = pp2_relevance_tokenize(
-        global_parameters.server->mergekey_pct, (const char *) mergekey);
-
-
-    norm_wr = wrbuf_alloc();
-    
-    while ((norm_str = pp2_relevance_token_next(prt)))
+    mergekey_norm = get_mergekey(xdoc, cl, record_no, service, se->nmem);
+    if (!mergekey_norm)
     {
-        if (*norm_str)
-        {
-            if (wrbuf_len(norm_wr))
-                wrbuf_puts(norm_wr, " ");
-            wrbuf_puts(norm_wr, norm_str);
-        }
+        yaz_log(YLOG_WARN, "Got no mergekey");
+        xmlFreeDoc(xdoc);
+        return 0;
     }
-        
-    mergekey_norm = (xmlChar *)nmem_strdup(se->nmem, wrbuf_cstr(norm_wr));
-    wrbuf_destroy(norm_wr);
-
-    pp2_relevance_token_destroy(prt);
+    record = record_create(se->nmem, 
+                           service->num_metadata, service->num_sortkeys, cl,
+                           record_no);
 
-    xmlFree(mergekey);
-    
     cluster = reclist_insert(se->reclist, 
-                             global_parameters.server->service, 
+                             service, 
                              record, (char *) mergekey_norm, 
                              &se->total_merged);
-    if (global_parameters.dump_records)
-        yaz_log(YLOG_LOG, "Cluster id %s from %s (#%d)", cluster->recid,
-                client_get_database(cl)->database->url, record_no);
     if (!cluster)
     {
         /* no room for record */
         xmlFreeDoc(xdoc);
         return 0;
     }
+    if (global_parameters.dump_records)
+        yaz_log(YLOG_LOG, "Cluster id %s from %s (#%d)", cluster->recid,
+                sdb->database->url, record_no);
     relevance_newrec(se->relevance, cluster);
-     
-     
-     // now parsing XML record and adding data to cluster or record metadata
-     for (n = root->children; n; n = n->next)
-     {
-         if (type)
-             xmlFree(type);
-         if (value)
-             xmlFree(value);
-         type = value = 0;
-
-         if (n->type != XML_ELEMENT_NODE)
-             continue;
-         if (!strcmp((const char *) n->name, "metadata"))
-         {
-             struct conf_metadata *ser_md = 0;
-             struct conf_sortkey *ser_sk = 0;
-             struct record_metadata **wheretoput = 0;
-             struct record_metadata *rec_md = 0;
-             int md_field_id = -1;
-             int sk_field_id = -1;
-
-             type = xmlGetProp(n, (xmlChar *) "type");
-             value = xmlNodeListGetString(xdoc, n->children, 1);
-
-             if (!type || !value || !*value)
-                 continue;
-
-             md_field_id 
-                 = conf_service_metadata_field_id(service, (const char *) type);
-             if (md_field_id < 0)
-             {
-                 yaz_log(YLOG_WARN, 
-                         "Ignoring unknown metadata element: %s", type);
-                 continue;
-             }
-
-             ser_md = &service->metadata[md_field_id];
-
-             if (ser_md->sortkey_offset >= 0){
-                 sk_field_id = ser_md->sortkey_offset;
-                 ser_sk = &service->sortkeys[sk_field_id];
-             }
-
-             // non-merged metadata
-             rec_md = record_metadata_init(se->nmem, (char *) value,
-                                           ser_md->type);
-             if (!rec_md)
-             {
-                 yaz_log(YLOG_WARN, "bad metadata data '%s' for element '%s'",
-                         value, type);
-                 continue;
-             }
-             wheretoput = &record->metadata[md_field_id];
-             while (*wheretoput)
-                 wheretoput = &(*wheretoput)->next;
-             *wheretoput = rec_md;
-
-             // merged metadata
-             rec_md = record_metadata_init(se->nmem, (char *) value,
-                                           ser_md->type);
-             wheretoput = &cluster->metadata[md_field_id];
-
-             // and polulate with data:
-             // assign cluster or record based on merge action
-             if (ser_md->merge == Metadata_merge_no)
-             {
-                 while (*wheretoput)
-                     wheretoput = &(*wheretoput)->next;
-                 *wheretoput = rec_md;
-             }
-             if (ser_md->merge == Metadata_merge_unique)
-             {
-                 struct record_metadata *mnode;
-                 for (mnode = *wheretoput; mnode; mnode = mnode->next)
-                     if (!strcmp((const char *) mnode->data.text.disp, 
-                                 rec_md->data.text.disp))
-                         break;
-                 if (!mnode)
-                 {
-                     rec_md->next = *wheretoput;
-                     *wheretoput = rec_md;
-                 }
-             }
-             else if (ser_md->merge == Metadata_merge_longest)
-             {
-                 if (!*wheretoput 
-                     || strlen(rec_md->data.text.disp) 
-                     > strlen((*wheretoput)->data.text.disp))
-                 {
-                     *wheretoput = rec_md;
-                     if (ser_sk)
-                     {
-                         const char *sort_str = 0;
-                         int skip_article = 
-                             ser_sk->type == Metadata_sortkey_skiparticle;
-
-                         if (!cluster->sortkeys[sk_field_id])
-                             cluster->sortkeys[sk_field_id] = 
-                                 nmem_malloc(se->nmem, 
-                                             sizeof(union data_types));
+    
+    // now parsing XML record and adding data to cluster or record metadata
+    for (n = root->children; n; n = n->next)
+    {
+        pp2_relevance_token_t prt;
+        if (type)
+            xmlFree(type);
+        if (value)
+            xmlFree(value);
+        type = value = 0;
+        
+        if (n->type != XML_ELEMENT_NODE)
+            continue;
+        if (!strcmp((const char *) n->name, "metadata"))
+        {
+            struct conf_metadata *ser_md = 0;
+            struct conf_sortkey *ser_sk = 0;
+            struct record_metadata **wheretoput = 0;
+            struct record_metadata *rec_md = 0;
+            int md_field_id = -1;
+            int sk_field_id = -1;
+            
+            type = xmlGetProp(n, (xmlChar *) "type");
+            value = xmlNodeListGetString(xdoc, n->children, 1);
+            
+            if (!type || !value || !*value)
+                continue;
+            
+            md_field_id 
+                = conf_service_metadata_field_id(service, (const char *) type);
+            if (md_field_id < 0)
+            {
+                if (se->number_of_warnings_unknown_metadata == 0)
+                {
+                    yaz_log(YLOG_WARN, 
+                            "Ignoring unknown metadata element: %s", type);
+                }
+                se->number_of_warnings_unknown_metadata++;
+                continue;
+            }
+            
+            ser_md = &service->metadata[md_field_id];
+            
+            if (ser_md->sortkey_offset >= 0){
+                sk_field_id = ser_md->sortkey_offset;
+                ser_sk = &service->sortkeys[sk_field_id];
+            }
+
+            // non-merged metadata
+            rec_md = record_metadata_init(se->nmem, (char *) value,
+                                          ser_md->type);
+            if (!rec_md)
+            {
+                yaz_log(YLOG_WARN, "bad metadata data '%s' for element '%s'",
+                        value, type);
+                continue;
+            }
+            wheretoput = &record->metadata[md_field_id];
+            while (*wheretoput)
+                wheretoput = &(*wheretoput)->next;
+            *wheretoput = rec_md;
+
+            // merged metadata
+            rec_md = record_metadata_init(se->nmem, (char *) value,
+                                          ser_md->type);
+            wheretoput = &cluster->metadata[md_field_id];
+
+            // and polulate with data:
+            // assign cluster or record based on merge action
+            if (ser_md->merge == Metadata_merge_unique)
+            {
+                struct record_metadata *mnode;
+                for (mnode = *wheretoput; mnode; mnode = mnode->next)
+                    if (!strcmp((const char *) mnode->data.text.disp, 
+                                rec_md->data.text.disp))
+                        break;
+                if (!mnode)
+                {
+                    rec_md->next = *wheretoput;
+                    *wheretoput = rec_md;
+                }
+            }
+            else if (ser_md->merge == Metadata_merge_longest)
+            {
+                if (!*wheretoput 
+                    || strlen(rec_md->data.text.disp) 
+                    > strlen((*wheretoput)->data.text.disp))
+                {
+                    *wheretoput = rec_md;
+                    if (ser_sk)
+                    {
+                        const char *sort_str = 0;
+                        int skip_article = 
+                            ser_sk->type == Metadata_sortkey_skiparticle;
+
+                        if (!cluster->sortkeys[sk_field_id])
+                            cluster->sortkeys[sk_field_id] = 
+                                nmem_malloc(se->nmem, 
+                                            sizeof(union data_types));
                          
-                         prt = pp2_relevance_tokenize(
-                             global_parameters.server->sort_pct,
-                             rec_md->data.text.disp);
+                        prt = pp2_relevance_tokenize(
+                            service->sort_pct,
+                            rec_md->data.text.disp);
 
-                         pp2_relevance_token_next(prt);
+                        pp2_relevance_token_next(prt);
                          
-                         sort_str = pp2_get_sort(prt, skip_article);
+                        sort_str = pp2_get_sort(prt, skip_article);
                          
-                         cluster->sortkeys[sk_field_id]->text.disp = 
-                             rec_md->data.text.disp;
-                         if (!sort_str)
-                         {
-                             sort_str = rec_md->data.text.disp;
-                             yaz_log(YLOG_WARN, 
-                                     "Could not make sortkey. Bug #1858");
-                         }
-                         cluster->sortkeys[sk_field_id]->text.sort = 
-                             nmem_strdup(se->nmem, sort_str);
+                        cluster->sortkeys[sk_field_id]->text.disp = 
+                            rec_md->data.text.disp;
+                        if (!sort_str)
+                        {
+                            sort_str = rec_md->data.text.disp;
+                            yaz_log(YLOG_WARN, 
+                                    "Could not make sortkey. Bug #1858");
+                        }
+                        cluster->sortkeys[sk_field_id]->text.sort = 
+                            nmem_strdup(se->nmem, sort_str);
 #if 0
-                         yaz_log(YLOG_LOG, "text disp=%s",
-                                 cluster->sortkeys[sk_field_id]->text.disp);
-                         yaz_log(YLOG_LOG, "text sort=%s",
-                                 cluster->sortkeys[sk_field_id]->text.sort);
+                        yaz_log(YLOG_LOG, "text disp=%s",
+                                cluster->sortkeys[sk_field_id]->text.disp);
+                        yaz_log(YLOG_LOG, "text sort=%s",
+                                cluster->sortkeys[sk_field_id]->text.sort);
 #endif
-                         pp2_relevance_token_destroy(prt);
+                        pp2_relevance_token_destroy(prt);
                     }
                 }
             }
@@ -1204,13 +1285,13 @@ struct record *ingest_record(struct client *cl, const char *rec,
             xmlFree(value);
             type = value = 0;
         }
-         else
-         {
-             if (se->number_of_warnings_unknown_elements == 0)
-                 yaz_log(YLOG_WARN,
-                         "Unexpected element in internal record: %s", n->name);
-             se->number_of_warnings_unknown_elements++;
-         }
+        else
+        {
+            if (se->number_of_warnings_unknown_elements == 0)
+                yaz_log(YLOG_WARN,
+                        "Unexpected element in internal record: %s", n->name);
+            se->number_of_warnings_unknown_elements++;
+        }
     }
     if (type)
         xmlFree(type);
@@ -1230,7 +1311,9 @@ struct record *ingest_record(struct client *cl, const char *rec,
 /*
  * Local variables:
  * c-basic-offset: 4
+ * c-file-style: "Stroustrup"
  * indent-tabs-mode: nil
  * End:
  * vim: shiftwidth=4 tabstop=8 expandtab
  */
+