Added Channel.setElementsToRead

README.md

@@ -56,7 +56,7 @@ Add the following dependency:
 <dependency>
   <groupId>org.epics</groupId>
   <artifactId>ca</artifactId>
-  <version>1.3.2</version>
+  <version>1.4.0</version>
 </dependency>
 ```
 
@@ -76,7 +76,7 @@ and the following repository:
 Add the following dependency:
 
 ```gradle
-compile 'org.epics:ca:1.3.2'
+compile 'org.epics:ca:1.4.0'
 ```
 
 and the following repository:

build.gradle

@@ -23,7 +23,7 @@ println "The Java version being used for this gradle build is: ${JavaVersion.cur
 
 // Define the coordinates for the build products.
 group = 'org.epics'
-version = '1.3.2'
+version = '1.4.0'
 
 // Note: currently (2020-06-01) the inspection error flagged up by IntelliJ is incorrect and has been suppressed.
 //noinspection GroovyUnusedAssignment

src/main/java/org/epics/ca/Channel.java

@@ -95,6 +95,15 @@ default  <MT extends Metadata<T>> Monitor<MT> addMonitor( Class<? extends Metada
    // get channel properties, e.g. native type, host, etc.
    Map<String, Object> getProperties();
 
+   //Number of elements to read.
+   //Default: 0 (defined by type or protocol)
+   //If negative then requests the native element count.
+   void setElementsToRead( int elementsToRead );
+
+   int getElementsToRead();
+
+   int getNativeElementCount();
+
    // suppresses AutoCloseable.close() exception
    @Override
    void close();

src/main/java/org/epics/ca/impl/ChannelImpl.java

@@ -6,6 +6,7 @@
 
 import org.apache.commons.lang3.Validate;
 import org.epics.ca.*;
+import org.epics.ca.annotation.CaChannel;
 import org.epics.ca.data.Metadata;
 import org.epics.ca.impl.TypeSupports.TypeSupport;
 import org.epics.ca.impl.monitor.MonitorNotificationService;
@@ -18,8 +19,10 @@
 
 import java.lang.reflect.Array;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -76,6 +79,8 @@ public class ChannelImpl<T> implements Channel<T>, TransportClient
 
    private volatile int nativeElementCount = 0;
 
+   private volatile int elementsToRead = 0;
+
    // on every connection loss the value gets incremented
    private final AtomicInteger connectionLossId = new AtomicInteger ();
 
@@ -374,6 +379,23 @@ public Map<String, Object> getProperties()
       return properties;
    }
 
+   @Override
+   public void setElementsToRead( int elementsToRead )
+   {
+      this.elementsToRead = elementsToRead;
+   }
+
+   @Override
+   public int getElementsToRead()
+   {
+      return getElementsToRead(typeSupport);
+   }
+
+   @Override
+   public int getNativeElementCount()
+   {
+      return nativeElementCount;
+   }
 
 /*- Public non-interface methods ---------------------------------------------*/
 
@@ -395,9 +417,17 @@ public synchronized TcpTransport getTcpTransport()
       return tcpTransport;
    }
 
-   public int getNativeElementCount()
+   public int getElementsToRead(TypeSupport typeSupport)
    {
-      return nativeElementCount;
+      int ret = (elementsToRead < 0) ?  nativeElementCount : elementsToRead;
+      ret = Math.max(Math.min(ret,nativeElementCount),0);
+      if(ret <=0) {
+         ret = typeSupport.getForcedElementCount();
+         if (ret == 0 && getTcpTransport().getMinorRevision() < 13) {
+            ret = nativeElementCount;
+         }
+      }
+      return ret;
    }
 
    /**

src/main/java/org/epics/ca/impl/requests/MonitorRequest.java

@@ -142,13 +142,7 @@ public void cancel()
 
    public void resubscribe( Transport transport )
    {
-      int dataCount = typeSupport.getForcedElementCount ();
-
-      if ( dataCount == 0 && channel.getTcpTransport().getMinorRevision () < 13 )
-      {
-         dataCount = channel.getNativeElementCount();
-      }
-
+      final int dataCount = channel.getElementsToRead(typeSupport);
       Messages.createSubscriptionMessage ( transport, typeSupport.getDataType (), dataCount, channel.getSID (), ioid, mask );
       transport.flush ();
    }
@@ -198,12 +192,7 @@ public void close()
          return;
       }
 
-      int dataCount = typeSupport.getForcedElementCount();
-
-      if ( dataCount == 0 && channel.getTcpTransport().getMinorRevision () < 13 )
-      {
-         dataCount = channel.getNativeElementCount();
-      }
+      final int dataCount = channel.getElementsToRead(typeSupport);
 
       try
       {

src/main/java/org/epics/ca/impl/requests/ReadNotifyRequest.java

@@ -60,11 +60,7 @@ public ReadNotifyRequest( ChannelImpl<?> channel, Transport transport, int sid,
       this.sid = sid;
       this.typeSupport = typeSupport;
 
-      final int minorRevision = channel.getTcpTransport().getMinorRevision();
-      final int forcedElementCount = typeSupport.getForcedElementCount();
-      final int nativeElementCount = channel.getNativeElementCount();
-
-      final int dataCount = ( forcedElementCount == 0 ) && ( minorRevision < 13 ) ? nativeElementCount : forcedElementCount;
+      final int dataCount = channel.getElementsToRead(typeSupport);
       logger.finest( "Receive data count is: " + dataCount );
       context = transport.getContext();
       ioid = context.registerResponseRequest( this );

src/test/java/org/epics/ca/LibraryVersionTest.java

@@ -42,7 +42,7 @@ void testLibraryVersion()
          // Note: the version string will have to be updated on every new release.
          // However, new releases should not be published until all the unit tests pass.
          logger.info( "Test is running from JAR. Version is: "  + versionString + "'." );
-         assertThat( LibraryVersion.getAsString(), is("1.3.2" ) );
+         assertThat( LibraryVersion.getAsString(), is("1.4.0" ) );
       }
       else
       {