Merge pull request #2284 from ControlSystemStudio/pva_beacons

PVA client search update

Author: kasemir

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

@@ -162,7 +162,7 @@ private void removeOldBeaconInfo(final Instant now)
         {
             final BeaconInfo info = infos.next().getValue();
             final long age = Duration.between(info.last, now).getSeconds();
-            if (age > 180) // TODO beacon_cleanup_period
+            if (age > PVASettings.EPICS_PVA_MAX_BEACON_AGE)
             {
                 logger.log(Level.FINER,
                            () -> "Removing beacon info " + info.guid + " (" + info.address + "), last seen " + age + " seconds ago");

core/pva/src/main/java/org/epics/pva/client/ChannelSearch.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/ClientUDPHandler.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/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
@@ -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/common/SearchRequest.java

@@ -35,6 +35,8 @@
 @SuppressWarnings("nls")
 public class PVAServer implements AutoCloseable
 {
+    // TODO Implement beacons?
+
     /** Common thread pool */
     public static ForkJoinPool POOL = ForkJoinPool.commonPool();
 

core/pva/src/main/java/org/epics/pva/server/PVAServer.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);
     }
 }

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

@@ -15,6 +15,9 @@
 import java.net.StandardProtocolFamily;
 import java.nio.ByteBuffer;
 import java.nio.channels.DatagramChannel;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 import java.util.logging.Level;
 
 import org.epics.pva.PVASettings;
@@ -176,24 +179,33 @@ private boolean handleSearch(final InetSocketAddress from, final byte version,
         if (search == null)
             return false;
 
-        if (search.name == null)
+        if (search.channels == null)
         {
             if (search.reply_required)
             {   // pvlist request
                 final boolean handled = server.handleSearchRequest(0, -1, null, search.client, null);
                 if (! handled  &&  search.unicast)
-                    PVAServer.POOL.submit(() -> forwardSearchRequest(0, -1, null, search.client));
+                    PVAServer.POOL.submit(() -> forwardSearchRequest(0, null, search.client));
             }
         }
         else
         {   // Channel search request
-            for (int i=0; i<search.name.length; ++i)
+            List<SearchRequest.Channel> forward = null;
+            for (SearchRequest.Channel channel : search.channels)
             {
-                final int cid = search.cid[i];
-                final String name = search.name[i];
-                final boolean handled = server.handleSearchRequest(search.seq, cid, name, search.client, null);
+                final boolean handled = server.handleSearchRequest(search.seq, channel.getCID(), channel.getName(), search.client, null);
                 if (! handled && search.unicast)
-                    PVAServer.POOL.submit(() -> forwardSearchRequest(search.seq, cid, name, search.client));
+                {
+                    if (forward == null)
+                        forward = new ArrayList<>();
+                    forward.add(channel);
+                }
+            }
+
+            if (forward != null)
+            {
+                final List<SearchRequest.Channel> to_forward = forward;
+                PVAServer.POOL.submit(() -> forwardSearchRequest(search.seq, to_forward, search.client));
             }
         }
 
@@ -209,19 +221,18 @@ private boolean handleSearch(final InetSocketAddress from, final byte version,
      *  allowing all servers on this host to reply.
      *
      *  @param seq Search sequence or 0
-     *  @param cid Channel ID or -1
-     *  @param name Name or <code>null</code>
+     *  @param channels Channel CIDs and names or <code>null</code> for 'list'
      *  @param address Client's address and port
      */
-    private void forwardSearchRequest(final int seq, final int cid, final String name, final InetSocketAddress address)
+    private void forwardSearchRequest(final int seq, final Collection<SearchRequest.Channel> channels, final InetSocketAddress address)
     {
         // TODO Remove the local IPv4 multicast re-send from the protocol, just use multicast from the start as with IPv6
         if (local_multicast == null)
             return;
         synchronized (send_buffer)
         {
             send_buffer.clear();
-            SearchRequest.encode(false, seq, cid, name, address, send_buffer);
+            SearchRequest.encode(false, seq, channels, address, send_buffer);
             send_buffer.flip();
             logger.log(Level.FINER, () -> "Forward search to " + local_multicast + "\n" + Hexdump.toHexdump(send_buffer));
             try