PVA client: Send multiple name searches in one message

Author: kasemir

core/pva/src/main/java/org/epics/pva/client/ChannelSearch.java

@@ -11,8 +11,8 @@
 
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
-import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Random;
@@ -217,7 +217,7 @@ public void start()
      */
     public void register(final PVAChannel channel, final boolean now)
     {
-        logger.log(Level.FINE, () -> "Register search for " + channel.getName() + " " + channel.getCID());
+        logger.log(Level.FINE, () -> "Register search for " + channel);
 
         final ClientChannelState old = channel.setState(ClientChannelState.SEARCHING);
         if (old == ClientChannelState.SEARCHING)
@@ -288,6 +288,7 @@ public void boost()
     private final ArrayList<PVAChannel> to_search = new ArrayList<>();
 
     /** Invoked by timer: Check searched channels for the next one to handle */
+    @SuppressWarnings("unchecked")
     private void runSearches()
     {
         to_search.clear();
@@ -369,9 +370,10 @@ else if (count == 0)
                 break;
 
             final List<PVAChannel> batch = to_search.subList(start, start + count);
-            System.out.println(Instant.now() + " Search bucket " + current_search_bucket.get() + " " + batch);
-            for (PVAChannel channel : batch)
-                search(channel);
+            logger.log(Level.FINE, "Search bucket " + current_search_bucket.get() + " " + batch);
+
+            // PVAChannel extends SearchRequest.Channel, so use List<PVAChannel> as Collection<SR.Channel>
+            search((Collection<SearchRequest.Channel>) (List<? extends SearchRequest.Channel>)batch);
             start += count;
         }
     }
@@ -385,22 +387,18 @@ public void list()
         synchronized (send_buffer)
         {
             logger.log(Level.FINE, "List Request");
-            sendSearch(0, -1, null);
+            sendSearch(0, null);
         }
     }
 
-    /** Issue search for channel
-     *  @param channel Channel to search
+    /** Issue search for channels
+     *  @param channels Channels to search, <code>null</code> for 'list'
      */
-    private void search(final PVAChannel channel)
+    private void search(final Collection<SearchRequest.Channel> channels)
     {
         // Search via TCP
         for (AddressInfo name_server : name_server_addresses)
         {
-            final SearchedChannel searched = searched_channels.get(channel.getCID());
-            if (searched == null)
-                continue;
-
             final ClientTCPHandler tcp = tcp_provider.apply(name_server.getAddress());
 
             // In case of connection errors (TCP connection blocked by firewall),
@@ -409,7 +407,7 @@ private void search(final PVAChannel channel)
             {
                 final RequestEncoder search_request = (version, buffer) ->
                 {
-                    logger.log(Level.FINE, () -> "Searching for " + channel + " via TCP " + tcp.getRemoteAddress());
+                    logger.log(Level.FINE, () -> "Searching for " + channels + " via TCP " + tcp.getRemoteAddress());
 
                     // Search sequence identifies the potentially repeated UDP.
                     // TCP search is once only, so PVXS always sends 0x66696E64 = "find".
@@ -419,7 +417,7 @@ private void search(final PVAChannel channel)
                     // Use 'any' reply address since reply will be via this TCP socket
                     final InetSocketAddress response_address = new InetSocketAddress(0);
 
-                    SearchRequest.encode(true, seq, channel.getCID(), channel.getName(), response_address , buffer);
+                    SearchRequest.encode(true, seq, channels, response_address , buffer);
                 };
                 tcp.submit(search_request);
             }
@@ -437,20 +435,20 @@ private void search(final PVAChannel channel)
             // For UDP, use bucket index to get a changing number that helps
             // match up duplicate packets and allows debugging bucket usage
             final int seq = current_search_bucket.get();
-            logger.log(Level.FINE, "UDP Search Request #" + seq + " for " + channel);
-            sendSearch(seq, channel.getCID(), channel.getName());
+            logger.log(Level.FINE, () -> "UDP Search Request #" + seq + " for " + channels);
+            sendSearch(seq, channels);
         }
     }
 
     /** Send a 'list' or channel search out via UDP */
-    private void sendSearch(final int seq, final int cid, final String name)
+    private void sendSearch(final int seq, final Collection<SearchRequest.Channel> channels)
     {
         // Buffer starts out with UNICAST bit set in the search message
         for (AddressInfo addr : unicast_search_addresses)
         {
             send_buffer.clear();
             final InetSocketAddress response = udp.getResponseAddress(addr);
-            SearchRequest.encode(true, seq, cid, name, response, send_buffer);
+            SearchRequest.encode(true, seq, channels, response, send_buffer);
             send_buffer.flip();
             try
             {
@@ -468,7 +466,7 @@ private void sendSearch(final int seq, final int cid, final String name)
         {
             send_buffer.clear();
             final InetSocketAddress response = udp.getResponseAddress(addr);
-            SearchRequest.encode(false, seq, cid, name, response, send_buffer);
+            SearchRequest.encode(false, seq, channels, response, send_buffer);
             send_buffer.flip();
             try
             {

core/pva/src/main/java/org/epics/pva/client/ClientUDPHandler.java

@@ -278,26 +278,25 @@ private boolean handleSearchRequest(final InetSocketAddress from, final byte ver
         {
             if (local_multicast != null  &&  search != null  &&  search.unicast)
             {
-                if (search.name == null)
+                if (search.channels == null)
                 {
                     if (search.reply_required)
                     {
                         forward_buffer.clear();
-                        SearchRequest.encode(false, 0, -1, null, search.client, forward_buffer);
+                        SearchRequest.encode(false, 0, null, search.client, forward_buffer);
                         forward_buffer.flip();
                         logger.log(Level.FINER, () -> "Forward search to list servers to " + local_multicast + "\n" + Hexdump.toHexdump(forward_buffer));
                         send(forward_buffer, local_multicast);
                     }
                 }
                 else
-                    for (int i=0; i<search.name.length; ++i)
-                    {
-                        forward_buffer.clear();
-                        SearchRequest.encode(false, search.seq, search.cid[i], search.name[i], search.client, forward_buffer);
-                        forward_buffer.flip();
-                        logger.log(Level.FINER, () -> "Forward search to " + local_multicast + "\n" + Hexdump.toHexdump(forward_buffer));
-                        send(forward_buffer, local_multicast);
-                    }
+                {
+                    forward_buffer.clear();
+                    SearchRequest.encode(false, search.seq, search.channels, search.client, forward_buffer);
+                    forward_buffer.flip();
+                    logger.log(Level.FINER, () -> "Forward search to " + local_multicast + "\n" + Hexdump.toHexdump(forward_buffer));
+                    send(forward_buffer, local_multicast);
+                }
             }
         }
         catch (Exception ex)

core/pva/src/main/java/org/epics/pva/client/PVAChannel.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2019-2021 Oak Ridge National Laboratory.
+ * Copyright (c) 2019-2022 Oak Ridge National Laboratory.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -17,6 +17,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 
+import org.epics.pva.common.SearchRequest;
 import org.epics.pva.data.PVADouble;
 import org.epics.pva.data.PVAString;
 import org.epics.pva.data.PVAStructure;
@@ -41,7 +42,7 @@
  *  @author Kay Kasemir
  */
 @SuppressWarnings("nls")
-public class PVAChannel implements AutoCloseable
+public class PVAChannel extends SearchRequest.Channel implements AutoCloseable
 {
     /** Provider for the 'next' client channel ID
      *
@@ -53,9 +54,7 @@ public class PVAChannel implements AutoCloseable
     private static final AtomicInteger CID_Provider = new AtomicInteger(1);
 
     private final PVAClient client;
-    private final String name;
     private final ClientChannelListener listener;
-    private final int cid = CID_Provider.incrementAndGet();
     private volatile int sid = -1;
 
     /** State
@@ -74,8 +73,8 @@ public class PVAChannel implements AutoCloseable
 
     PVAChannel(final PVAClient client, final String name, final ClientChannelListener listener)
     {
+        super(CID_Provider.incrementAndGet(), name);
         this.client = client;
-        this.name = name;
         this.listener = listener;
     }
 
@@ -92,25 +91,12 @@ ClientTCPHandler getTCP() throws Exception
         return copy;
     }
 
-    /** @return Client channel ID */
-    int getCID()
-    {
-        return cid;
-    }
-
     /** @return Server channel ID */
     int getSID()
     {
         return sid;
     }
 
-
-    /** @return Channel name */
-    public String getName()
-    {
-        return name;
-    }
-
     /** @return {@link ClientChannelState} */
     public ClientChannelState getState()
     {

core/pva/src/main/java/org/epics/pva/common/SearchRequest.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2019-2021 Oak Ridge National Laboratory.
+ * Copyright (c) 2019-2022 Oak Ridge National Laboratory.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -12,6 +12,9 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 import java.util.logging.Level;
 
 import org.epics.pva.data.PVAAddress;
@@ -23,6 +26,43 @@
 @SuppressWarnings("nls")
 public class SearchRequest
 {
+    /** Channel with CID to be searched */
+    public static class Channel
+    {
+        /** Client ID */
+        protected final int cid;
+
+        /** Channel name */
+        protected final String name;
+
+        /** @param cid Client ID
+         *  @param name Channel name
+         */
+        public Channel(final int cid, final String name)
+        {
+            this.cid = cid;
+            this.name = name;
+        }
+
+        /** @return Client channel ID */
+        public int getCID()
+        {
+            return cid;
+        }
+
+        /** @return Channel name */
+        public String getName()
+        {
+            return name;
+        }
+
+        @Override
+        public String toString()
+        {
+            return "'" + name + "' [CID " + cid + "]";
+        }
+    };
+
     /** Sequence number */
     public int seq;
     /** Is it a unicast? */
@@ -31,10 +71,8 @@ public class SearchRequest
     public boolean reply_required;
     /** Address of client */
     public InetSocketAddress client;
-    /** Names requested in search */
-    public String[] name;
-    /** Client IDs for names */
-    public int[] cid;
+    /** Names requested in search, <code>null</code> for 'list' */
+    public List<Channel> channels;
 
     /** Check search request
      *
@@ -105,7 +143,7 @@ public static SearchRequest decode(final InetSocketAddress from, final byte vers
 
         if (count == 0)
         {   // pvlist request
-            search.name = null;
+            search.channels = null;
             logger.log(Level.FINER, () -> "PVA Client " + from + " sent search #" + search.seq + " to list servers");
         }
         else
@@ -115,15 +153,13 @@ public static SearchRequest decode(final InetSocketAddress from, final byte vers
                 logger.log(Level.WARNING, "PVA Client " + from + " sent search #" + search.seq + " for protocol '" + protocol + "', need 'tcp'");
                 return null;
             }
-            search.name = new String[count];
-            search.cid  = new int[count];
+            search.channels = new ArrayList<>(count);
             for (int i=0; i<count; ++i)
             {
                 final int cid = buffer.getInt();
                 final String name = PVAString.decodeString(buffer);
                 logger.log(Level.FINER, () -> "PVA Client " + from + " sent search #" + search.seq + " for " + name + " [cid " + cid + "]");
-                search.name[i] = name;
-                search.cid[i] = cid;
+                search.channels.add(new Channel(cid, name));
             }
         }
 
@@ -132,12 +168,11 @@ public static SearchRequest decode(final InetSocketAddress from, final byte vers
 
     /** @param unicast Unicast?
      *  @param seq Sequence number
-     *  @param cid Client ID
-     *  @param name PV name
+     *  @param channels Channels to search, <code>null</code> for 'list'
      *  @param address client's address
      *  @param buffer Buffer into which to encode
      */
-    public static void encode(final boolean unicast, final int seq, final int cid, final String name, final InetSocketAddress address, final ByteBuffer buffer)
+    public static void encode(final boolean unicast, final int seq, final Collection<Channel> channels, final InetSocketAddress address, final ByteBuffer buffer)
     {
         // Create with zero payload size, to be patched later
         PVAHeader.encodeMessageHeader(buffer, PVAHeader.FLAG_NONE, PVAHeader.CMD_SEARCH, 0);
@@ -152,7 +187,7 @@ public static void encode(final boolean unicast, final int seq, final int cid, f
         // Mark search message as unicast so that receiver will forward
         // it via local broadcast to other local listeners.
         // 0-bit for replyRequired, 7-th bit for "sent as unicast" (1)/"sent as broadcast/multicast" (0)
-        buffer.put((byte) ((unicast ? 0x80 : 0x00) | (cid < 0 ? 0x01 : 0x00)));
+        buffer.put((byte) ((unicast ? 0x80 : 0x00) | (channels == null ? 0x01 : 0x00)));
 
         // reserved
         buffer.put((byte) 0);
@@ -166,8 +201,9 @@ public static void encode(final boolean unicast, final int seq, final int cid, f
         buffer.putShort((short)address.getPort());
 
         // string[] protocols with count as byte since < 254
-        // struct { int searchInstanceID, string channelName } channels[] with count as short?!
-        if (cid < 0)
+        // struct { int searchInstanceID, string channelName } channels[] with count as short
+        // No protocol and empty channels[] for 'list' aka 'discover' request
+        if (channels == null)
         {
             buffer.put((byte)0);
             buffer.putShort((short)0);
@@ -177,9 +213,12 @@ public static void encode(final boolean unicast, final int seq, final int cid, f
             buffer.put((byte)1);
             PVAString.encodeString("tcp", buffer);
 
-            buffer.putShort((short)1);
-            buffer.putInt(cid);
-            PVAString.encodeString(name, buffer);
+            buffer.putShort((short)channels.size());
+            for (Channel channel : channels)
+            {
+                buffer.putInt(channel.cid);
+                PVAString.encodeString(channel.name, buffer);
+            }
         }
 
         // Update payload size

core/pva/src/main/java/org/epics/pva/server/SearchCommandHandler.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2021 Oak Ridge National Laboratory.
+ * Copyright (c) 2021-2022 Oak Ridge National Laboratory.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -32,9 +32,9 @@ public void handleCommand(final ServerTCPHandler tcp, final ByteBuffer buffer) t
 
         final SearchRequest search = SearchRequest.decode(tcp.getRemoteAddress(), version, payload_size, buffer);
 
-        if (search.name != null)
-            for (int i=0; i<search.name.length; ++i)
-                tcp.getServer().handleSearchRequest(search.seq, search.cid[i], search.name[i],
+        if (search.channels != null)
+            for (SearchRequest.Channel channel : search.channels)
+                tcp.getServer().handleSearchRequest(search.seq, channel.getCID(), channel.getName(),
                                                     search.client, tcp);
     }
 }