RpcSearchInvoker extends SearchInvoker implements Client.ResponseReceiver { @Override protected Object sendSearchRequest(Query query, Object incomingContext) { this.query = query; Client.NodeConnection nodeConnection = resourcePool.getConnection(node.key()); if (nodeConnection == null) { responses.add(Client.ResponseOrError.fromError("Could not send search to unknown node " + node.key())); responseAvailable(); return incomingContext; } query.trace(false, 5, "Sending search request with jrt/protobuf to node with dist key ", node.key()); RpcContext context = getContext(incomingContext); double timeoutSeconds = ((double) query.getTimeLeft() - 3.0) / 1000.0; nodeConnection.request(RPC_METHOD, context.compressedPayload.type(), context.compressedPayload.uncompressedSize(), context.compressedPayload.data(), this, timeoutSeconds); return context; } RpcSearchInvoker(VespaBackEndSearcher searcher, Node node, RpcResourcePool resourcePool, int maxHits); void receive(Client.ResponseOrError<ProtobufResponse> response);  }
@Test public void testProtobufSerialization() throws IOException { var compressionTypeHolder = new AtomicReference<CompressionType>(); var payloadHolder = new AtomicReference<byte[]>(); var lengthHolder = new AtomicInteger(); var mockClient = parameterCollectorClient(compressionTypeHolder, payloadHolder, lengthHolder); var mockPool = new RpcResourcePool(ImmutableMap.of(7, mockClient.createConnection("foo", 123))); var invoker = new RpcSearchInvoker(mockSearcher(), new Node(7, "seven", 1), mockPool, 1000); Query q = new Query("search/?query=test&hits=10&offset=3"); RpcSearchInvoker.RpcContext context = (RpcSearchInvoker.RpcContext) invoker.sendSearchRequest(q, null); assertEquals(lengthHolder.get(), context.compressedPayload.uncompressedSize()); assertSame(context.compressedPayload.data(), payloadHolder.get()); var bytes = mockPool.compressor().decompress(payloadHolder.get(), compressionTypeHolder.get(), lengthHolder.get()); var request = SearchProtocol.SearchRequest.newBuilder().mergeFrom(bytes).build(); assertEquals(10, request.getHits()); assertEquals(3, request.getOffset()); assertTrue(request.getQueryTreeBlob().size() > 0); var invoker2 = new RpcSearchInvoker(mockSearcher(), new Node(8, "eight", 1), mockPool, 1000); RpcSearchInvoker.RpcContext context2 = (RpcSearchInvoker.RpcContext)invoker2.sendSearchRequest(q, context); assertSame(context, context2); assertEquals(lengthHolder.get(), context.compressedPayload.uncompressedSize()); assertSame(context.compressedPayload.data(), payloadHolder.get()); }
@Test public void testProtobufSerializationWithMaxHitsSet() throws IOException { int maxHits = 5; var compressionTypeHolder = new AtomicReference<CompressionType>(); var payloadHolder = new AtomicReference<byte[]>(); var lengthHolder = new AtomicInteger(); var mockClient = parameterCollectorClient(compressionTypeHolder, payloadHolder, lengthHolder); var mockPool = new RpcResourcePool(ImmutableMap.of(7, mockClient.createConnection("foo", 123))); var invoker = new RpcSearchInvoker(mockSearcher(), new Node(7, "seven", 1), mockPool, maxHits); Query q = new Query("search/?query=test&hits=10&offset=3"); invoker.sendSearchRequest(q, null); var bytes = mockPool.compressor().decompress(payloadHolder.get(), compressionTypeHolder.get(), lengthHolder.get()); var request = SearchProtocol.SearchRequest.newBuilder().mergeFrom(bytes).build(); assertEquals(maxHits, request.getHits()); }