Use an injected VespaDocumentAccess for streaming search

4 files changed, 45 insertions, 70 deletions

diff --git a/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java b/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java
index ccb6e1248b4..307b3e17373 100644
--- a/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java
+++ b/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java
@@ -1,11 +1,13 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.prelude.cluster;
 
+import com.google.inject.Inject;
 import com.yahoo.component.ComponentId;
 import com.yahoo.component.chain.dependencies.After;
 import com.yahoo.component.provider.ComponentRegistry;
 import com.yahoo.container.QrConfig;
 import com.yahoo.container.QrSearchersConfig;
+import com.yahoo.container.core.documentapi.VespaDocumentAccess;
 import com.yahoo.container.handler.VipStatus;
 import com.yahoo.prelude.IndexFacts;
 import com.yahoo.prelude.fastsearch.ClusterParams;
@@ -61,13 +63,14 @@ public class ClusterSearcher extends Searcher {
 
     private VespaBackEndSearcher server = null;
 
+    @Inject
     public ClusterSearcher(ComponentId id,
                            QrSearchersConfig qrsConfig,
                            ClusterConfig clusterConfig,
                            DocumentdbInfoConfig documentDbConfig,
                            ComponentRegistry<Dispatcher> dispatchers,
                            QrConfig qrConfig,
-                           VipStatus vipStatus) {
+                           VespaDocumentAccess access) {
         super(id);
 
         int searchClusterIndex = clusterConfig.clusterId();
@@ -93,7 +96,7 @@ public class ClusterSearcher extends Searcher {
 
         if (searchClusterConfig.indexingmode() == STREAMING) {
             VdsStreamingSearcher searcher = vdsCluster(qrConfig.discriminator(), searchClusterIndex,
-                                                       searchClusterConfig, docSumParams, documentDbConfig);
+                                                       searchClusterConfig, docSumParams, documentDbConfig, access);
             addBackendSearcher(searcher);
             vipStatus.addToRotation(searcher.getName());
         } else {
@@ -139,13 +142,14 @@ public class ClusterSearcher extends Searcher {
                                                    int searchclusterIndex,
                                                    QrSearchersConfig.Searchcluster searchClusterConfig,
                                                    SummaryParameters docSumParams,
-                                                   DocumentdbInfoConfig documentdbInfoConfig) {
+                                                   DocumentdbInfoConfig documentdbInfoConfig,
+                                                   VespaDocumentAccess access) {
         if (searchClusterConfig.searchdef().size() != 1) {
             throw new IllegalArgumentException("Search clusters in streaming search shall only contain a single searchdefinition : " + searchClusterConfig.searchdef());
         }
         ClusterParams clusterParams = makeClusterParams(searchclusterIndex);
-        VdsStreamingSearcher searcher = new VdsStreamingSearcher();
-        searcher.setSearchClusterConfigId(searchClusterConfig.rankprofiles().configid());
+        VdsStreamingSearcher searcher = new VdsStreamingSearcher(access);
+        searcher.setSearchClusterName(searchClusterConfig.rankprofiles().configid());
         searcher.setDocumentType(searchClusterConfig.searchdef(0));
         searcher.setStorageClusterRouteSpec(searchClusterConfig.storagecluster().routespec());
         searcher.init(serverId, docSumParams, clusterParams, documentdbInfoConfig);
diff --git a/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsStreamingSearcher.java b/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsStreamingSearcher.java
index 3528da17dfe..3a4ff4d385b 100644
--- a/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsStreamingSearcher.java
+++ b/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsStreamingSearcher.java
@@ -1,9 +1,13 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.streamingvisitors;
 
+import com.yahoo.container.core.documentapi.VespaDocumentAccess;
 import com.yahoo.document.DocumentId;
 import com.yahoo.document.select.parser.ParseException;
 import com.yahoo.document.select.parser.TokenMgrException;
+import com.yahoo.documentapi.VisitorParameters;
+import com.yahoo.documentapi.VisitorSession;
+import com.yahoo.documentapi.messagebus.loadtypes.LoadTypeSet;
 import com.yahoo.fs4.DocsumPacket;
 import com.yahoo.messagebus.routing.Route;
 import com.yahoo.prelude.Ping;
@@ -53,15 +57,15 @@ public class VdsStreamingSearcher extends VespaBackEndSearcher {
 
     private Route route;
     /** The configId used to access the searchcluster. */
-    private String searchClusterConfigId = null;
+    private String searchClusterName = null;
     private String documentType;
     /** The route to the storage cluster. */
     private String storageClusterRouteSpec = null;
 
-    private String getSearchClusterConfigId() { return searchClusterConfigId; }
+    private String getSearchClusterName() { return searchClusterName; }
     private String getStorageClusterRouteSpec() { return storageClusterRouteSpec; }
-    public final void setSearchClusterConfigId(String clusterName) {
-        this.searchClusterConfigId = clusterName;
+    public final void setSearchClusterName(String clusterName) {
+        this.searchClusterName = clusterName;
     }
     public final void setDocumentType(String documentType) {
         this.documentType = documentType;
@@ -71,16 +75,35 @@ public class VdsStreamingSearcher extends VespaBackEndSearcher {
         this.storageClusterRouteSpec = storageClusterRouteSpec;
     }
 
-    private static class VdsVisitorFactory implements VisitorFactory {
+    private static class VespaVisitorFactory implements VdsVisitor.VisitorSessionFactory, VisitorFactory {
+
+        private final VespaDocumentAccess access;
+
+        private VespaVisitorFactory(VespaDocumentAccess access) {
+            this.access = access;
+        }
+
+        @Override
+        public VisitorSession createVisitorSession(VisitorParameters params) throws ParseException {
+            return access.createVisitorSession(params);
+        }
+
+        @Override
+        public LoadTypeSet getLoadTypeSet() {
+            return access.delegate().getParams().getLoadTypes();
+        }
+
         @Override
         public Visitor createVisitor(Query query, String searchCluster, Route route, String documentType, int traceLevelOverride) {
-            return new VdsVisitor(query, searchCluster, route, documentType, traceLevelOverride);
+            return new VdsVisitor(query, searchCluster, route, documentType, this, traceLevelOverride);
         }
+
     }
 
-    public VdsStreamingSearcher() {
-        this(new VdsVisitorFactory());
+    public VdsStreamingSearcher(VespaDocumentAccess access) {
+        this(new VespaVisitorFactory(access));
     }
+
     VdsStreamingSearcher(VisitorFactory visitorFactory) {
         this.visitorFactory = visitorFactory;
         tracingOptions = TracingOptions.DEFAULT;
@@ -146,11 +169,11 @@ public class VdsStreamingSearcher extends VespaBackEndSearcher {
                     "only one of these query parameters to be set: streaming.userid, streaming.groupname, " +
                     "streaming.selection"));
         }
-        query.trace("Routing to search cluster " + getSearchClusterConfigId() + " and document type " + documentType, 4);
+        query.trace("Routing to search cluster " + getSearchClusterName() + " and document type " + documentType, 4);
         long timeStartedNanos = tracingOptions.getClock().nanoTimeNow();
         int effectiveTraceLevel = inferEffectiveQueryTraceLevel(query);
 
-        Visitor visitor = visitorFactory.createVisitor(query, getSearchClusterConfigId(), route, documentType, effectiveTraceLevel);
+        Visitor visitor = visitorFactory.createVisitor(query, getSearchClusterName(), route, documentType, effectiveTraceLevel);
         try {
             visitor.doSearch();
         } catch (ParseException e) {
diff --git a/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsVisitor.java b/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsVisitor.java
index 7d7b78f7153..49fda880b44 100644
--- a/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsVisitor.java
+++ b/container-search/src/main/java/com/yahoo/vespa/streamingvisitors/VdsVisitor.java
@@ -83,55 +83,6 @@ class VdsVisitor extends VisitorDataHandler implements Visitor {
         LoadTypeSet getLoadTypeSet();
     }
 
-    private static class MessageBusVisitorSessionFactory implements VisitorSessionFactory {
-        private static final Object initMonitor = new Object();
-        private static final AtomicReference<MessageBusVisitorSessionFactory> instance = new AtomicReference<>();
-
-        private final LoadTypeSet loadTypes;
-        private final DocumentAccess access;
-
-        private MessageBusVisitorSessionFactory() {
-            loadTypes = new LoadTypeSet("client");
-            access = new MessageBusDocumentAccess(new MessageBusParams(loadTypes));
-        }
-
-        @Override
-        public VisitorSession createVisitorSession(VisitorParameters params) throws ParseException {
-            return access.createVisitorSession(params);
-        }
-
-        @Override
-        public LoadTypeSet getLoadTypeSet() {
-            return loadTypes;
-        }
-
-        /**
-         * Returns a single, shared instance of this class which is lazily created in a thread-safe
-         * manner the first time this method is invoked.
-         *
-         * May throw any config-related exception if subscription fails.
-         */
-        static MessageBusVisitorSessionFactory sharedInstance() {
-            var ref = instance.getAcquire();
-            if (ref != null) {
-                return ref;
-            }
-            synchronized (initMonitor) {
-                ref = instance.getAcquire();
-                if (ref != null) {
-                    return ref;
-                }
-                ref = new MessageBusVisitorSessionFactory();
-                instance.setRelease(ref);
-            }
-            return ref;
-        }
-    }
-
-    public VdsVisitor(Query query, String searchCluster, Route route, String documentType, int traceLevelOverride) {
-        this(query, searchCluster, route, documentType, MessageBusVisitorSessionFactory.sharedInstance(), traceLevelOverride);
-    }
-
     public VdsVisitor(Query query, String searchCluster, Route route,
                       String documentType, VisitorSessionFactory visitorSessionFactory,
                       int traceLevelOverride)
@@ -264,9 +215,8 @@ class VdsVisitor extends VisitorDataHandler implements Visitor {
     static int getQueryFlags(Query query) {
         int flags = 0;
 
-        boolean requestCoverage=true; // Always request coverage information
+        boolean requestCoverage = true; // Always request coverage information
 
-        flags |= 0; // was collapse
         flags |= query.properties().getBoolean(Model.ESTIMATE) ? 0x00000080 : 0;
         flags |= (query.getRanking().getFreshness() != null) ? 0x00002000 : 0;
         flags |= requestCoverage ? 0x00008000 : 0;
@@ -344,9 +294,7 @@ class VdsVisitor extends VisitorDataHandler implements Visitor {
         }
 
         if (params.getControlHandler().getResult().code == VisitorControlHandler.CompletionCode.SUCCESS) {
-            if (log.isLoggable(Level.FINE)) {
-                log.log(Level.FINE, "VdsVisitor completed successfully for " + query + " with selection " + params.getDocumentSelection());
-            }
+            log.log(Level.FINE, () -> "VdsVisitor completed successfully for " + query + " with selection " + params.getDocumentSelection());
         } else {
             throw new IllegalArgumentException("Query failed: " +
                                                params.getControlHandler().getResult().code + ": " +
diff --git a/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java b/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java
index f4608f1c991..465d54ae050 100644
--- a/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java
+++ b/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java
@@ -530,7 +530,7 @@ public class ClusterSearcherTestCase {
                                    documentDbConfig.build(),
                                    dispatchers,
                                    new QrConfig.Builder().build(),
-                                   vipStatus);
+                                   null);
     }
 
     private static ClusterInfoConfig createClusterInfoConfig() {