[Java] Tidy up after merge of PR #684.

Author: mjpt777

aeron-cluster/src/main/java/io/aeron/cluster/ConsensusModule.java

@@ -139,20 +139,20 @@ public static State get(final AtomicCounter counter)
         }
 
         /**
-         * Get the {@link State} corresponding to a particular value.
+         * Get the {@link State} corresponding to a particular code.
          *
-         * @param value of the State.
-         * @return the {@link State} corresponding to the provided value.
-         * @throws ClusterException if the value does not correspond to a valid State.
+         * @param code representing a {@link State}.
+         * @return the {@link State} corresponding to the provided code.
+         * @throws ClusterException if the code does not correspond to a valid State.
          */
-        public static State get(final int value)
+        public static State get(final int code)
         {
-            if (value < 0 || value > (STATES.length - 1))
+            if (code < 0 || code > (STATES.length - 1))
             {
-                throw new ClusterException("invalid state counter code: " + value);
+                throw new ClusterException("invalid state counter code: " + code);
             }
 
-            return STATES[value];
+            return STATES[code];
         }
     }
 

aeron-cluster/src/main/java/io/aeron/cluster/ConsensusModuleAgent.java

@@ -615,7 +615,6 @@ public void onSnapshotRecordings(final long correlationId, final SnapshotRecordi
         }
     }
 
-    @SuppressWarnings("unused")
     public void onJoinCluster(final long leadershipTermId, final int memberId)
     {
         final ClusterMember member = clusterMemberByIdMap.get(memberId);

aeron-samples/src/main/java/io/aeron/samples/AeronStat.java

@@ -160,11 +160,7 @@ public static void main(final String[] args) throws Exception
         final CncFileReader cncFileReader = CncFileReader.map();
 
         final CounterFilter counterFilter = new CounterFilter(
-            typeFilter,
-            identityFilter,
-            sessionFilter,
-            streamFilter,
-            channelFilter);
+            typeFilter, identityFilter, sessionFilter, streamFilter, channelFilter);
 
         if (watch)
         {
@@ -176,17 +172,15 @@ public static void main(final String[] args) throws Exception
         }
     }
 
-    private static void workLoop(final long delayMs, final Runnable printOutput) throws Exception
+    private static void workLoop(final long delayMs, final Runnable outputPrinter) throws Exception
     {
         final AtomicBoolean running = new AtomicBoolean(true);
         SigInt.register(() -> running.set(false));
 
         do
         {
             clearScreen();
-
-            printOutput.run();
-
+            outputPrinter.run();
             Thread.sleep(delayMs);
         }
         while (running.get());
@@ -200,7 +194,7 @@ private static void printOutput(final CncFileReader cncFileReader, final Counter
         System.out.println(
             " - Aeron Stat (CnC v" + cncFileReader.semanticVersion() + ")" +
             ", pid " + SystemUtil.getPid() +
-            ", heartbeat " + cncFileReader.driverHeartbeatAge() + "ms");
+            ", heartbeat age " + cncFileReader.driverHeartbeatAgeMs() + "ms");
         System.out.println("======================================================================");
 
         final CountersReader counters = cncFileReader.countersReader();
@@ -296,9 +290,9 @@ else if ((typeId >= PUBLISHER_LIMIT_TYPE_ID && typeId <= RECEIVER_POS_TYPE_ID) |
             {
                 return
                     match(identityFilter, () -> Long.toString(keyBuffer.getLong(REGISTRATION_ID_OFFSET))) &&
-                        match(sessionFilter, () -> Integer.toString(keyBuffer.getInt(SESSION_ID_OFFSET))) &&
-                        match(streamFilter, () -> Integer.toString(keyBuffer.getInt(STREAM_ID_OFFSET))) &&
-                        match(channelFilter, () -> keyBuffer.getStringAscii(CHANNEL_OFFSET));
+                    match(sessionFilter, () -> Integer.toString(keyBuffer.getInt(SESSION_ID_OFFSET))) &&
+                    match(streamFilter, () -> Integer.toString(keyBuffer.getInt(STREAM_ID_OFFSET))) &&
+                    match(channelFilter, () -> keyBuffer.getStringAscii(CHANNEL_OFFSET));
             }
             else if (typeId >= SEND_CHANNEL_STATUS_TYPE_ID && typeId <= RECEIVE_CHANNEL_STATUS_TYPE_ID)
             {

aeron-samples/src/main/java/io/aeron/samples/CncFileReader.java

@@ -1,3 +1,18 @@
+/*
+ * Copyright 2014-2019 Real Logic Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.aeron.samples;
 
 import io.aeron.CncFileDescriptor;
@@ -6,7 +21,8 @@
 import org.agrona.DirectBuffer;
 import org.agrona.IoUtil;
 import org.agrona.SemanticVersion;
-import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+import org.agrona.concurrent.ringbuffer.RingBufferDescriptor;
 import org.agrona.concurrent.status.CountersReader;
 
 import java.io.File;
@@ -16,15 +32,16 @@
 import static io.aeron.samples.SamplesUtil.mapExistingFileReadOnly;
 
 /**
- * A utility class for interpreting the cnc file.
+ * Reader for Aeron CnC file represented by {@link CncFileDescriptor} which can be used for observability.
  */
 public final class CncFileReader implements AutoCloseable
 {
-    private final MappedByteBuffer cncByteBuffer;
+    private boolean isClosed = false;
     private final int cncVersion;
-    private final CountersReader countersReader;
-    private final ManyToOneRingBuffer toDriverBuffer;
     private final String cncSemanticVersion;
+    private final MappedByteBuffer cncByteBuffer;
+    private final CountersReader countersReader;
+    private final UnsafeBuffer toDriverBuffer;
 
     private CncFileReader(final MappedByteBuffer cncByteBuffer)
     {
@@ -46,8 +63,7 @@ private CncFileReader(final MappedByteBuffer cncByteBuffer)
         this.cncVersion = cncVersion;
         this.cncSemanticVersion = SemanticVersion.toString(cncVersion);
 
-        this.toDriverBuffer = new ManyToOneRingBuffer(
-            CncFileDescriptor.createToDriverBuffer(cncByteBuffer, cncMetaDataBuffer));
+        this.toDriverBuffer = CncFileDescriptor.createToDriverBuffer(cncByteBuffer, cncMetaDataBuffer);
 
         this.countersReader = new CountersReader(
             createCountersMetaDataBuffer(cncByteBuffer, cncMetaDataBuffer),
@@ -64,6 +80,7 @@ public static CncFileReader map()
     {
         final File cncFile = CommonContext.newDefaultCncFile();
         final MappedByteBuffer cncByteBuffer = mapExistingFileReadOnly(cncFile);
+
         return new CncFileReader(cncByteBuffer);
     }
 
@@ -98,28 +115,34 @@ public CountersReader countersReader()
     }
 
     /**
-     * Get the timestamp (ms) of the last driver heartbeat.
+     * Get the epoch timestamp (ms) of the last driver heartbeat.
      *
-     * @return the timestamp (ms) of the last driver heartbeat.
+     * @return the epoch timestamp (ms) of the last driver heartbeat.
      */
-    public long driverHeartbeat()
+    public long driverHeartbeatMs()
     {
-        return toDriverBuffer.consumerHeartbeatTime();
+        final int timestampOffset = (toDriverBuffer.capacity() - RingBufferDescriptor.TRAILER_LENGTH) +
+            RingBufferDescriptor.CONSUMER_HEARTBEAT_OFFSET;
+
+        return toDriverBuffer.getLongVolatile(timestampOffset);
     }
 
     /**
      * Get the number of milliseconds since the last driver heartbeat.
      *
      * @return the number of milliseconds since the last driver heartbeat.
      */
-    public long driverHeartbeatAge()
+    public long driverHeartbeatAgeMs()
     {
-        return System.currentTimeMillis() - driverHeartbeat();
+        return System.currentTimeMillis() - driverHeartbeatMs();
     }
 
-    @Override
     public void close()
     {
-        IoUtil.unmap(cncByteBuffer);
+        if (!isClosed)
+        {
+            isClosed = true;
+            IoUtil.unmap(cncByteBuffer);
+        }
     }
 }