public interface Channel<T> extends AutoCloseable
{
   String getName();

   ConnectionState getConnectionState();

   AccessRights getAccessRights();

   Channel<T> connect();

   CompletableFuture<Channel<T>> connectAsync();


   //
   // listeners
   //

   Listener addConnectionListener( BiConsumer<Channel<T>, Boolean> handler );

   Listener addAccessRightListener( BiConsumer<Channel<T>, AccessRights> handler );


   //
   // sync methods, exception is thrown on failure
   //

   // NOTE: reusable get methods
   //	T get(T reuse);
   // CompletableFuture<T> getAsync(T reuse);
   // <MT extends Metadata<T>> MT get(Class<? extends Metadata> clazz, T reuse);
   // <MT extends Metadata<T>> CompletableFuture<MT> getAsync(Class<? extends MT> clazz, T reuse);

   T get();

   void put( T value );

   void putNoWait( T value ); // best-effort put

   //
   // async methods, exception is reported via CompletableFuture
   //
   CompletableFuture<T> getAsync();

   CompletableFuture<Status> putAsync( T value );

   // NOTE: "public <MT extends Metadata<T>> MT get(Class<MT> clazz)" would
   // be a better definition, however it raises unchecked warnings in the code
   // and requires explicit casts for monitor APIs
   // the drawback of the signature below is that type of "?" can be different than "MT"
   // (it will raise ClassCastException if not properly used)

   @SuppressWarnings( "rawtypes" )
   <MT extends Metadata<T>> MT get( Class<? extends Metadata> clazz );

   @SuppressWarnings( "rawtypes" )
   <MT extends Metadata<T>> CompletableFuture<MT> getAsync( Class<? extends Metadata> clazz );

   //
   // monitors
   //

   // Value-only monitor. Default, value-change, notification mask.
   default Monitor<T> addValueMonitor( Consumer<? super T> handler )
   {
      return addValueMonitor (handler, Monitor.VALUE_MASK);
   }

   // Value-only monitor. User-specified notification mask.
   Monitor<T> addValueMonitor( Consumer<? super T> handler, int mask );

   // Metadata monitor.  Default, value-change, notification mask.
   @SuppressWarnings( "rawtypes" )
   default  <MT extends Metadata<T>> Monitor<MT> addMonitor( Class<? extends Metadata> clazz, Consumer<MT> handler )
   {
      return addMonitor( clazz, handler, Monitor.VALUE_MASK );
   }

   // Metadata monitor.  User-specified notification mask.
   @SuppressWarnings( "rawtypes" )
   <MT extends Metadata<T>> Monitor<MT> addMonitor( Class<? extends Metadata> clazz, Consumer<MT> handler, int mask );

   //
   // misc
   //
   // get channel properties, e.g. native type, host, etc.
   Map<String, Object> getProperties();

   // suppresses AutoCloseable.close() exception
   @Override
   void close();
} public class Constants
{

/*- Public attributes --------------------------------------------------------*/

   public enum ChannelProperties
   {
      nativeType, nativeTypeCode, remoteAddress, nativeElementCount
   }

   /**
    * String value of the JVM property key which specifies whether to strip the hostname returned by
    * InetAddress.getLocalHost().getHostName().
    */
   public static final String CA_STRIP_HOSTNAME = "CA_STRIP_HOSTNAME";

   /**
    * String value of the JVM property key to provide (override) hostname.
    */
   public static final String CA_HOSTNAME_KEY = "HOSTNAME";

   /**
    * Minimal priority.
    */
   public static final short CHANNEL_PRIORITY_MIN = 0;

   /**
    * Maximal priority.
    */
   public static final short CHANNEL_PRIORITY_MAX = 99;

   /**
    * Default priority.
    */
   public static final short CHANNEL_PRIORITY_DEFAULT = CHANNEL_PRIORITY_MIN;

   /**
    * DB links priority.
    */
   public static final short CHANNEL_PRIORITY_LINKS_DB = CHANNEL_PRIORITY_MAX;

   /**
    * Archive priority.
    */
   public static final short CHANNEL_PRIORITY_ARCHIVE = (short) ((CHANNEL_PRIORITY_MAX + CHANNEL_PRIORITY_MIN) / 2);

   /**
    * OPI priority.
    */
   public static final short CHANNEL_PRIORITY_OPI = CHANNEL_PRIORITY_MIN;

   /* -------- Core CA constants -------- */

   /**
    * CA protocol major revision (implemented by this library).
    */
   public static final short CA_MAJOR_PROTOCOL_REVISION = 4;

   /**
    * CA protocol minor revision (implemented by this library).
    */
   public static final short CA_MINOR_PROTOCOL_REVISION = 13;

   /**
    * Unknown CA protocol minor revision.
    */
   public static final short CA_UNKNOWN_MINOR_PROTOCOL_REVISION = 0;

   /**
    * Initial delay in milliseconds between creation of a new Context and the
    * first attempt to register with the CA Repeater.
    */
   public static final int CA_REPEATER_INITIAL_REGISTRATION_DELAY = 500;

   /**
    * CA Repeater attempted registration interval in milliseconds.
    */
   public static final int CA_REPEATER_REGISTRATION_INTERVAL = 60_000;

   /**
    * CA protocol message header size.
    */
   public static final short CA_MESSAGE_HEADER_SIZE = 16;

   /**
    * CA protocol message extended header size.
    */
   public static final short CA_EXTENDED_MESSAGE_HEADER_SIZE = (short) (CA_MESSAGE_HEADER_SIZE + 8);

   /**
    * UDP maximum send message size.
    * MAX_UDP: 1500 (max of Ethernet and 802.{2,3} MTU) - 20(IP) - 8(UDP)
    * (the MTU of Ethernet is currently independent of its speed variant)
    */
   public static final int MAX_UDP_SEND = 1024;

   /**
    * UDP maximum receive message size.
    */
   public static final int MAX_UDP_RECV = 0xFFFF + 16;

   /**
    * TCP maximum receive message size.
    */
   public static final int MAX_TCP_RECV = 1024 * 16 + CA_EXTENDED_MESSAGE_HEADER_SIZE;

   /**
    * Default priority (corresponds to POSIX SCHED_OTHER)
    */
   public static final short CA_DEFAULT_PRIORITY = 0;

   /**
    * Read access right mask.
    */
   public static final int CA_PROTO_ACCESS_RIGHT_READ = 1;

   /**
    * Write access right mask.
    */
   public static final int CA_PROTO_ACCESS_RIGHT_WRITE = 1 << 1;

   /**
    * Do not require response for CA search request.
    */
   public static final short CA_SEARCH_DONTREPLY = 5;

   /**
    * Require response (even if not found) for CA search request over TCP.
    */
   public static final short CA_SEARCH_DOREPLY = 10;

   /**
    * Echo (state-of-health message) response timeout in ms.
    */
   public static final long CA_ECHO_TIMEOUT = 5000;

   /**
    * Max. (requested) string size.
    */
   public static final int MAX_STRING_SIZE = 40;

   /**
    * Unreasonable channel name length.
    */
   public static final int UNREASONABLE_CHANNEL_NAME_LENGTH = 500;

/*- Private attributes -------------------------------------------------------*/
/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/
/*- Private methods ----------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/
/*- Private methods ----------------------------------------------------------*/
/*- Nested Classes -----------------------------------------------------------*/

} public class MonitorNotificationServiceFactoryCreator implements AutoCloseable
{

/*- Public attributes --------------------------------------------------------*/

   enum ServiceImpl
   {
      BlockingQueueSingleWorkerMonitorNotificationServiceImpl,
      BlockingQueueMultipleWorkerMonitorNotificationServiceImpl,
      StripedExecutorServiceMonitorNotificationServiceImpl;

      private static final ServiceImpl[] copyOfValues = values();

      public static boolean isRecognised( String name )
      {
         for ( ServiceImpl value : copyOfValues )
         {
            if (value.name().equals(name))
            {
               return true;
            }
         }
         return false;
      }
   }

   /**
    * This definition configures the behaviour of the CA library V_1_1_0 release.
    *
    * @implNote
    * This release used a BlockingQueueMonitorNotificationService with 5 queue workers.
    */
   public static final String V1_1_0_DEFAULT_IMPL = ServiceImpl.BlockingQueueMultipleWorkerMonitorNotificationServiceImpl.name() + ",5";

   /**
    * This definition configures the behaviour that is applicable for a wide range of clients.
    *
    * @implNote
    * The current implementation uses a BlockingQueueMultipleWorkerMonitorNotificationService running with 16 notification threads.
    * This may change in future releases.
    */
   public static final String DEFAULT_IMPL = ServiceImpl.BlockingQueueMultipleWorkerMonitorNotificationServiceImpl.name() + ",16";

   /**
    * This definition configures the behaviour for a client that does not mind dropping intermediate events to keep up
    * with the notification rate.
    *
    * @implNote
    * The current implementation uses a BlockingQueueMultipleWorkerMonitorNotificationServiceImpl running with 100
    * notification threads and a non-buffering queue.
    *
    * This may change in future releases.
    */
   public static final String HUMAN_CONSUMER_IMPL = ServiceImpl.BlockingQueueMultipleWorkerMonitorNotificationServiceImpl + ",100,1";

   /**
    * This definition configures the behaviour for a client which requires events to be buffered to keep up with the
    * notification rate during busy periods.
    *
    * @implNote
    * The current implementation uses a StripedExecutorMonitorNotificationService running with 100 notification threads.
    * This may change in future releases.
    */
   public static final String MACHINE_CONSUMER_IMPL = ServiceImpl.BlockingQueueMultipleWorkerMonitorNotificationServiceImpl.name() + ",100";

   /**
    * The number of service threads that will be used by default for service implementations which require
    * more than one thread.
    */
   public static final int NUMBER_OF_SERVICE_THREADS_DEFAULT = 10;

   /**
    * The size of the notification value buffer which will be used by default.
    *
    * @implNote
    * This definition currently applies to the BlockingQueueSingleWorkerMonitorNotificationServiceImpl,
    * BlockingQueueMultipleWorkerMonitorNotificationServiceImpl service implementations.
    */
   public static final int NOTIFICATION_VALUE_BUFFER_SIZE_DEFAULT = Integer.MAX_VALUE;

/*- Private attributes -------------------------------------------------------*/

   private static final Logger logger = LibraryLogManager.getLogger( MonitorNotificationServiceFactoryCreator.class );
   private static final List<MonitorNotificationServiceFactory> serviceFactoryList = new ArrayList<>();


/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/

   /**
    * Constructs a factory which will generate service instances based on the
    * specified service implementation.
    *
    * The following properties are supported:
    * <ul>
    * <li> BlockingQueueSingleWorkerMonitorNotificationServiceImpl,XXXX[,BufferSize]</li>
    * <li> BlockingQueueMultipleWorkerMonitorNotificationServiceImpl[,NumberOfThreads][,BufferSize]</li>
    * <li> StripedExecutorServiceMonitorNotificationServiceImpl[,NumberOfThreads]</li>
    * </ul>
    *
    * @param serviceConfiguration specifies the properties of the service instances that
    *    this factory will generate.
    *
    * @return the factory.
    *
    * @throws IllegalArgumentException if the serviceImpl string was of the
    * incorrect format.
    */
   public static MonitorNotificationServiceFactory create( String serviceConfiguration )
   {
      logger.finest( String.format( "MonitorNotificationServiceFactoryCreator create method has been called with serviceImpl specifier: %s", serviceConfiguration ) );

      Validate.notEmpty( serviceConfiguration );

      final String[] args = serviceConfiguration.split( "," );
      Validate.isTrue( args.length >= 1 );
      Validate.isTrue( ServiceImpl.isRecognised(args[ 0 ] ), "The service configuration '" + args[ 0 ] + "' was not recognised." );

      final ServiceImpl serviceImpl = ServiceImpl.valueOf(args[ 0 ] );

      final MonitorNotificationServiceFactory serviceFactory ;
      switch( serviceImpl )
      {
         case StripedExecutorServiceMonitorNotificationServiceImpl:
         {
            final int totalNumberOfServiceThreads = (args.length == 2) ? NumberUtils.toInt( args[ 1 ], NUMBER_OF_SERVICE_THREADS_DEFAULT) : NUMBER_OF_SERVICE_THREADS_DEFAULT;
            serviceFactory = new StripedExecutorServiceMonitorNotificationServiceFactory( totalNumberOfServiceThreads );
         }
         break;

         case BlockingQueueSingleWorkerMonitorNotificationServiceImpl:
         {
            if( (args.length >= 2) )
            {
               logger.info( "Note: in this implementation the value for the number of notification threads will be ignored and set to 1." );
            }

            final int totalNumberOfServiceThreads = 1;
            final int notificationValueBufferQueueSize = (args.length == 3) ? NumberUtils.toInt(args[ 2 ], NOTIFICATION_VALUE_BUFFER_SIZE_DEFAULT ) : NOTIFICATION_VALUE_BUFFER_SIZE_DEFAULT ;

            serviceFactory = new BlockingQueueMonitorNotificationServiceFactory( totalNumberOfServiceThreads, notificationValueBufferQueueSize );
         }
         break;

         default:
         case BlockingQueueMultipleWorkerMonitorNotificationServiceImpl:
         {
            final int totalNumberOfServiceThreads = (args.length >= 2) ? NumberUtils.toInt(args[ 1 ], NUMBER_OF_SERVICE_THREADS_DEFAULT) : NUMBER_OF_SERVICE_THREADS_DEFAULT;
            final int notificationValueBufferQueueSize = (args.length == 3) ? NumberUtils.toInt(args[ 2 ], NOTIFICATION_VALUE_BUFFER_SIZE_DEFAULT ) : NOTIFICATION_VALUE_BUFFER_SIZE_DEFAULT;

            serviceFactory = new BlockingQueueMonitorNotificationServiceFactory( totalNumberOfServiceThreads, notificationValueBufferQueueSize );
         }
         break;
      }
      serviceFactoryList.add( serviceFactory );
      return serviceFactory;
   }

/*- Public methods -----------------------------------------------------------*/

   @Override
   public void close()
   {
      logger.finer(  "MonitorNotificationServiceFactoryCreator close method has been called." );
      serviceFactoryList.forEach( MonitorNotificationServiceFactory::close );
      serviceFactoryList.clear();
      logger.finer( "MonitorNotificationServiceFactoryCreator close method completed." );

   }

   /**
    * Returns a list of all service implementations recognised by this factory.
    *
    * @return the list.
    */
   public static List<String> getAllServiceImplementations()
   {
      return Arrays.stream( ServiceImpl.values() ).map( ServiceImpl::toString ).collect( Collectors.toList() );
   }

   static public long getServiceCount()
   {
      return serviceFactoryList.stream().mapToInt( MonitorNotificationServiceFactory::getServiceCount ).sum();
   }

   /**
    * Utility method to cleanly shutdown an ExecutorService
    *
    * @param executorService the service to shut down.
    */
   public static void shutdownExecutor( ExecutorService executorService )
   {
      logger.fine( "Starting executor shutdown sequence for executor " + executorService );

      executorService.shutdown();
      try
      {
         logger.finer( "Waiting up to 2 seconds for tasks to finish..." );
         if ( executorService.awaitTermination(2, TimeUnit.SECONDS ) )
         {
            logger.finer( "Executor terminated ok." );
         }
         else
         {
            logger.finer( "Executor did not yet terminate. Forcing termination..." );
            final List<Runnable> droppedTasks = executorService.shutdownNow();
            logger.finer( "Executor was abruptly shutdown. " + droppedTasks.size() + " tasks will not be executed." );
            logger.finer( "Termination forced. Waiting up to 2 seconds for tasks to finish..." );
            executorService.awaitTermination(2, TimeUnit.SECONDS );
         }
      }
      catch ( InterruptedException ex )
      {
         logger.warning( "Interrupted whilst waiting for tasks to finish. Propagating interrupt." );
         Thread.currentThread().interrupt();
      }
      logger.fine( "Executor shutdown sequence completed." );
   }

/*- Private methods ----------------------------------------------------------*/
/*- Nested Classes -----------------------------------------------------------*/

} public interface MonitorNotificationServiceFactory extends AutoCloseable
{

/*- Class Declaration --------------------------------------------------------*/
/*- Public attributes --------------------------------------------------------*/
/*- Private attributes -------------------------------------------------------*/
/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/

   /**
    * Returns a service instance which will publish events to the specified Consumer.
    *
    * @param consumer the consumer to publish to.
    * @param <T> the type of events that this service instance will publish.
    *
    * @return the service instance.
    */
   <T> MonitorNotificationService<T> getServiceForConsumer( Consumer<? super T> consumer );

   /**
    * Closes down this service factory and releases all underlying resources.
    */
   @Override
   void close();

   /**
    * Returns the count of service instances created by this provider.
    *
    * @return the result.
    */
   int getServiceCount();

   /**
    * Returns an indication of whether this service implementation can be expected
    * to drop notifications or whether it will buffer them in situations where
    * the publication rate exceeds the rate at which the consumers can process the
    * data.
    *
    * @return the result.
    */
   boolean getQosMetricIsBuffered();

   /**
    * Returns an indication of the size of each consumer's notification buffer
    * (for service implementations which provide buffering).
    *
    * Returns 1 in the case that the service implementation is not buffered.
    *
    * @return the result.
    */
   int getQosMetricBufferSizePerConsumer();

   /**
    * Returns an indication of how many threads the consumer may be called back on.
    * Where multiple threads are involved the consumer may wish to synchronize their
    * accept method to force serialisation of the notification sequence so that
    * events get notified in the same sequence as they were published.
    *
    * @return the result.
    */
   int getQosMetricNumberOfNotificationThreadsPerConsumer();

   /**
    * Returns an indication of whether the publish method accepts null
    * as a valid token to be sent to the Coinsumer.
    *
    * @return the result.
    */
   boolean getQosMetricIsNullPublishable();


/*- Private methods ----------------------------------------------------------*/
/*- Nested Classes -----------------------------------------------------------*/

} public class Reactor
{

   // Get Logger
   private static final Logger logger = LibraryLogManager.getLogger( Reactor.class );

   /**
    * Simple internal request interface.
    */
   private interface InternalRequest
   {
      /**
       * Process the request.
       */
      void process();
   }

   /**
    * Registration request to selector.
    */
   private class RegistrationRequest implements InternalRequest
   {
      private final SelectableChannel selectableChannel;
      private final int interestOps;
      private final ReactorHandler handler;

      private SelectionKey key;
      private ClosedChannelException exception;
      private boolean done = false;

      /**
       * Constructor.
       *
       * @param selectableChannel the channel
       * @param interestOps the operation of interest.
       * @param handler the handler
       */
      public RegistrationRequest( SelectableChannel selectableChannel, int interestOps, ReactorHandler handler )
      {
         this.selectableChannel = selectableChannel;
         this.interestOps = interestOps;
         this.handler = handler;
      }

      /**
       * Process the registration resulting set <code>key</code> or <code>exception</code> fields.
       */
      public synchronized void process()
      {
         try
         {
            // do the registration...
            key = selectableChannel.register (selector, interestOps, handler);
         }
         catch ( ClosedChannelException cce )
         {
            exception = cce;
         }
         finally
         {
            // notify about completed registration process
            done = true;
            notifyAll ();
         }
      }

      /**
       * Cancel process of registration.
       */
      public synchronized void cancelRegistration()
      {
         // notify about canceled registration process
         done = true;
         notifyAll ();
      }

      /**
       * Exception thrown during registration, <code>null</code> if none.
       *
       * @return exception thrown during registration.
       */
      public ClosedChannelException getException()
      {
         return exception;
      }

      /**
       * Obtained key, <code>null</code> on failure.
       *
       * @return obtained key.
       */
      public SelectionKey getKey()
      {
         return key;
      }

      /**
       * Checks if registration is done (or canceled).
       * Note: note synced to this.
       *
       * @return true when done.
       */
      public boolean isDone()
      {
         return done;
      }

   }

   /**
    * Registration request to selector.
    */
   private static class DeregistrationRequest implements InternalRequest
   {
      private final SelectionKey key;

      /**
       * Constructor.
       *
       * @param key the key
       */
      public DeregistrationRequest( SelectionKey key )
      {
         this.key = key;
      }

      /**
       * Unregisters and closes.
       */
      public void process()
      {
         key.cancel ();
         try
         {
            key.channel ().close ();
         }
         catch ( IOException e )
         { /* noop */ }
      }
   }

   /**
    * Changing operation of interests request.
    */
   private static class InterestOpsChangeRequest implements InternalRequest
   {
      private final SelectionKey selectionKey;
      private final int interestOps;

      /**
       * @param selectionKey they key
       * @param interestOps the operation of interest
       */
      public InterestOpsChangeRequest( SelectionKey selectionKey, int interestOps )
      {
         this.selectionKey = selectionKey;
         this.interestOps = interestOps;
      }

      /**
       * Process change of interests ops.
       */
      public void process()
      {
         selectionKey.interestOps (interestOps);
      }
   }

   /**
    * IO selector (e.g. UNIX select() implementation).
    */
   private Selector selector;

   /**
    * Estimated (max) number of requests (aka connections).
    */
   private final int DEFAULT_ESTIMATED_MAX_REQUESTS = 1024;

   /**
    * List of pending registration request(s).
    */
   private final Deque<RegistrationRequest> registrationRequests = new ArrayDeque<> (DEFAULT_ESTIMATED_MAX_REQUESTS);

   /**
    * List of pending registration request(s).
    */
   private final Deque<DeregistrationRequest> unregistrationRequests = new ArrayDeque<> (DEFAULT_ESTIMATED_MAX_REQUESTS);

   /**
    * List of pending registration request(s).
    */
   private final Deque<InterestOpsChangeRequest> interestOpsChangeRequests = new ArrayDeque<> (DEFAULT_ESTIMATED_MAX_REQUESTS);

   /**
    * Map of disabled keys, pairs (SelectionKey key, Integer interestOps).
    */
   private final Map<SelectionKey, Integer> disabledKeys = new HashMap<> (DEFAULT_ESTIMATED_MAX_REQUESTS);

   /**
    * Selector status.
    */
   private final AtomicInteger selectorPending = new AtomicInteger (0);

   /**
    * Shutdown status.
    */
   private final AtomicBoolean shutdown = new AtomicBoolean( false );

   /**
    * Shutdown monitor (condition variable).
    */
   private final Object shutdownMonitor = new Object ();

   /**
    * Creates a new instance of reactor.
    *
    * @throws IOException IO Exception
    */
   public Reactor() throws IOException
   {
      initialize ();
   }

   /**
    * Initialize reactor.
    *
    * @throws IOException IO Exception
    */
   private void initialize() throws IOException
   {
      // create an instance of selector
      selector = Selector.open ();
   }

   /**
    * Processes internal request list.
    * Also takes care of sync.
    *
    * @param list list of internal requests to be processed.
    */
   private static void processInternalRequest( Deque<? extends InternalRequest> list )
   {
      if ( !list.isEmpty () )
      {

         // TODO: These days (2020) there are probably simpler synchronisation techniques available
         // TODO: that could clean up this code. But not without some decent unit tests which we currently
         // TODO: don't have.
         //noinspection SynchronizationOnLocalVariableOrMethodParameter
         synchronized ( list )
         {
            while ( !list.isEmpty () )
            {
               try
               {
                  list.removeFirst().process ();
               }
               catch ( Throwable th )
               {
                  // noop (just not to lose control)
               }
            }
         }
      }
   }

   /**
    * Process requests.
    * NOTE: this method has to be called until <code>false</code> is returned.
    *
    * @return <code>true</code> if selector is still active, <code>false</code> when shutdown.
    */
   public boolean process()
   {
      // do while reactor is open
      if ( selector.isOpen () && !shutdown.get() )
      {
         processInternal ();
      }

      // if closed, do the cleanup
      if ( !selector.isOpen () || shutdown.get() )
      {
         // cancel pending registration requests
         if ( !registrationRequests.isEmpty () )
         {
            synchronized ( registrationRequests )
            {
               while ( !registrationRequests.isEmpty () )
               {
                  registrationRequests.removeFirst().cancelRegistration();
               }
            }
         }

         // check pending unregistration requests (not to forget to close channels)
         processInternalRequest (unregistrationRequests);

         // TODO: These days (2020) there are probably simpler synchronisation techniques available
         // TODO: that could clean up this code. But not without some decent unit tests which we currently
         // TODO: don't have.
         synchronized ( shutdownMonitor )
         {
            shutdownMonitor.notifyAll ();
         }

         //System.out.println();
         //System.out.println("[Selector closed.]");

         return false;
      }
      else
         return true;
   }

   /**
    * Process requests (internal method).
    * NOTE: Selector objects are thread-safe, but the key sets they contain are not. The key sets
    * returned by the keys( ) and selectedKeys( ) methods are direct references to private
    * Set objects inside the Selector object. These sets can change at any time.
    */
   private void processInternal()
   {
      //System.err.println("[processInternal] " + Thread.currentThread().getName());

      try
      {

         int numSelectedKeys = 0;
         while ( numSelectedKeys == 0 && !shutdown.get() )
         {

            // check pending unregistration requests
            processInternalRequest (unregistrationRequests);

            // check pending interestOps change requests
            processInternalRequest (interestOpsChangeRequests);

            //System.err.println("[select] " + Thread.currentThread().getName());

            selectorPending.incrementAndGet ();
            try
            {
               // possible improvement: do the low-latency trick (busy-wait for a moment, and then sleep)

               // wait for selection, but only if necessary
               /* int */
               numSelectedKeys = selector.selectedKeys ().size ();
               if ( numSelectedKeys == 0 )
               {
                  numSelectedKeys = selector.select();
               }
            }
            finally
            {
               selectorPending.decrementAndGet ();
            }

            // check shutdown status
            if ( shutdown.get() )
            {
               selector.close ();
               return;
            }

            // check pending registration requests
            processInternalRequest (registrationRequests);

         }

         // wake-up or forced selectNow()
         if ( numSelectedKeys == 0 || shutdown.get() )
            return;

         //System.out.println();
         //System.out.println("[Selector selected # keys: " + numSelectedKeys + "]");

         Iterator<SelectionKey> selectedKeys = selector.selectedKeys ().iterator ();

         // process only one request per select (to support concurrent processing)
         SelectionKey selectedKey = selectedKeys.next ();

         // the definition of OP_WRITE in select agrees with the Unix definition, ie. not edge triggered like Win32
         // this means that you must add and remove OP_WRITE from the interestOps depending on the actual ability to write
         // clear SelectionKey.OP_WRITE here...
         int ops;
         try
         {
            ops = selectedKey.interestOps ();
            if ( (ops & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE )
               selectedKey.interestOps (ops & (~SelectionKey.OP_WRITE));
         }
         catch ( CancelledKeyException cke )
         {
            // noop
         }

         // get handler as attachment
         ReactorHandler handler = (ReactorHandler) selectedKey.attachment ();

         // request accepted, remove key from set
         // NOTE: this has to be done before processing (to support concurrent processing)
         selectedKeys.remove ();

         try
         {
            // process request
            handler.handleEvent (selectedKey);
         }
         catch ( CancelledKeyException cke )
         {
            // noop
         }

      }
      catch ( Throwable th )
      {
         logger.log (Level.SEVERE, "Unexpected exception caught while processing selection keys.", th);
      }

      //System.err.println("[processInternal done] " + Thread.currentThread().getName());
   }

   /**
    * Unregisters <code>SelectableChannel</code> from the reactor.
    *
    * @param selectableChannel channel to be unregistered.
    */
   public void unregisterAndClose( SelectableChannel selectableChannel )
   {
      SelectionKey key = selectableChannel.keyFor (selector);
      synchronized ( unregistrationRequests )
      {
         unregistrationRequests.add (new DeregistrationRequest(key));
      }
      selector.wakeup ();
   }

   /**
    * Registers <code>SelectableChannel</code> to the reactor.
    *
    * @param selectableChannel channel to be registered.
    * @param interestOps       operations supported by channel (i.e. operations of interest).
    * @param handler           handle to process requests.
    * @return selector selection key.
    * @throws ClosedChannelException Channel closed
    */
   @SuppressWarnings( "UnusedReturnValue" )
   public SelectionKey register( SelectableChannel selectableChannel, int interestOps, ReactorHandler handler ) throws ClosedChannelException
   {
      final RegistrationRequest rr = new RegistrationRequest (selectableChannel, interestOps, handler);

      // TODO: These days (2020) there are probably simpler synchronisation techniques available
      // TODO: that could clean up this code. But not without some decent unit tests which we currently
      // TODO: don't have.
      //noinspection SynchronizationOnLocalVariableOrMethodParameter
      synchronized ( rr )
      {
         // pend registration request...
         synchronized ( registrationRequests )
         {
            registrationRequests.add (rr);
         }

         // check for shutdown status
         // NOTE: this has to be done in synchronization block after registering request
         if ( shutdown.get() )
         {
            throw new IllegalStateException("Reactor is already shutdown.");
         }

         // ... and wake-up selector to have this registration valid immediately
         selector.wakeup ();

         // if already awake process request immediately,
         // this is race-condition proof, since wake-up guarantees
         // that thread will we awaken immediately and will not select
         if ( selectorPending.get () == 0 )
         {
            rr.process();
         }
         else
         {
            // wait for completion
            try
            {
               while ( !rr.isDone () )
               {
                  rr.wait();
               }
            }
            catch ( InterruptedException ie )
            { /* noop */ }
         }
      }

      // registration during shutdown
      if ( shutdown.get() )
      {
         throw new IllegalStateException("Reactor is shutting down.");
      }
      // exception occurred during registration
      else if ( rr.getException () != null )
      {
         throw rr.getException();
      }
      // return obtained key
      else
      {
         return rr.getKey();
      }
   }

   /**
    * Change <code>SelectionKey</code> operations of interest.
    *
    * @param selectionKey selection key
    * @param interestOps  interest ops
    */
   private void setInterestOpsInternal( SelectionKey selectionKey, int interestOps )
   {
      // NOTE: interestOps blocks in W2K if there is a select active on the key
      // selectionKey.interestOps(interestOps);
      synchronized ( interestOpsChangeRequests )
      {
         interestOpsChangeRequests.add (new InterestOpsChangeRequest(selectionKey, interestOps));
      }
      selector.wakeup ();
   }

   /**
    * Disable selection key - sets its interest ops to 0 and stores its current interest ops (to be restored).
    * NOTE: to be called only (otherwise it can block) when select is not pending
    *
    * @param key selection key to be disabled.
    */
   public void disableSelectionKey( SelectionKey key )
   {
      synchronized ( disabledKeys )
      {
         if ( disabledKeys.containsKey (key) )
            return;

         // save current operations of interest
         int interestOps = key.interestOps ();
         disabledKeys.put (key, interestOps);

         // disable the key
         // since this can be called only when select is not pending, we can do this
         key.interestOps (0);
      }
   }

   /**
    * Enable selection key - restore its interest ops.
    * NOTE: can be called when select is pending
    *
    * @param key selection key to be enabled.
    */
   public void enableSelectionKey( SelectionKey key )
   {
      synchronized ( disabledKeys )
      {
         Integer ops = disabledKeys.remove (key);
         if ( ops != null )
         {
            setInterestOpsInternal(key, ops);
         }
      }
   }

   /**
    * Change <code>SelectionKey</code> operations of interest.
    *
    * @param channel     channel
    * @param interestOps interest ops
    */
   public void setInterestOps( AbstractSelectableChannel channel, int interestOps )
   {
      SelectionKey key = channel.keyFor (selector);
      if ( key != null )
         setInterestOps (key, interestOps);
   }

   /**
    * Change <code>SelectionKey</code> operations of interest.
    * If channel is disabled, it changes its stored interest ops.
    *
    * @param selectionKey the key.
    * @param interestOps the operation of interest.
    */
   private void setInterestOps( SelectionKey selectionKey, int interestOps )
   {
      synchronized ( disabledKeys )
      {
         Integer ops = disabledKeys.get (selectionKey);
         if ( ops != null )
            disabledKeys.put (selectionKey, interestOps );
         else
            setInterestOpsInternal (selectionKey, interestOps);
      }
   }

   /**
    * Shutdown the reactor.
    */
   public void shutdown()
   {
      if ( shutdown.get() )
      {
         return;
      }

      // TODO: these days (2020) there are probably simpler synchronisation techniques available
      // TODO: that could clean up this code. But not without some decent unit tests which we currently
      // TODO: don't have.
      synchronized( shutdownMonitor )
      {
         shutdown.set( true );
         selector.wakeup ();

         try
         {
            shutdownMonitor.wait ();
         }
         catch ( InterruptedException e )
         { /* noop */ }

      }
   }

   /**
    * Get shutdown status.
    *
    * @return shutdown status.
    */
   public boolean isShutdown()
   {
      return shutdown.get();
   }
} public interface ReactorHandler
{

   /**
    * Process request of given <code>SelectionKey</code>.
    *
    * @param key key to be processed.
    */
   void handleEvent( SelectionKey key );

} public class LeaderFollowersHandler implements ReactorHandler, Runnable
{

   /**
    * Reactor to serve.
    */
   protected final Reactor reactor;

   /**
    * Application-specific event processing handler.
    */
   protected final ReactorHandler handler;

   /**
    * Leader-followers thread pool.
    */
   protected final LeaderFollowersThreadPool threadPool;

   /**
    * Constructor.
    *
    * @param reactor    reactor to handle.
    * @param handler    application-specific event processing handler.
    * @param threadPool leader-followers thread pool.
    */
   public LeaderFollowersHandler( Reactor reactor, ReactorHandler handler, LeaderFollowersThreadPool threadPool )
   {
      this.reactor = reactor;
      this.handler = handler;
      this.threadPool = threadPool;
   }

   /**
    * @see org.epics.ca.impl.reactor.ReactorHandler#handleEvent(java.nio.channels.SelectionKey)
    */
   public void handleEvent( SelectionKey key )
   {
      // if not valid, just promote new leader
      if ( !key.isValid () )
      {
         threadPool.promoteLeader (this);
         return;
      }

      reactor.disableSelectionKey (key);
      try
      {
         // promote a follower thread to become a leader
         threadPool.promoteLeader (this);
         //System.err.println("[processing] " + Thread.currentThread().getName());
         // dispatch application-specific event processing code
         handler.handleEvent( key );
      }
      finally
      {
         // enable the handle in the reactor
         reactor.enableSelectionKey(key);
      }
   }

   /**
    * @see java.lang.Runnable#run()
    */
   public void run()
   {
      reactor.process ();
   }

} public class LeaderFollowersThreadPool
{
   private static final Logger logger = LibraryLogManager.getLogger( LeaderFollowersThreadPool.class );

   /**
    * Default thread pool size.
    */
   public static final int DEFAULT_THREADPOOL_SIZE = 5;

   /**
    * Shutdown status flag.
    */
   private final AtomicBoolean shutdown = new AtomicBoolean( false );

   /**
    * Executor.
    */
   private final ThreadPoolExecutor executor;

   /**
    * Constructor.
    */
   public LeaderFollowersThreadPool()
   {

      int threadPoolSize = DEFAULT_THREADPOOL_SIZE;
      String strVal = System.getProperty (this.getClass ().getName () + ".thread_pool_size", String.valueOf (threadPoolSize));
      if ( strVal != null )
      {
         try
         {
            // minimum are two threads (leader and one follower)
            threadPoolSize = Math.max (2, Integer.parseInt (strVal));
         }
         catch ( NumberFormatException nfe )
         { /* noop */ }
      }

      // NOTE: consider using LIFO ordering of threads (to maximize CPU cache affinity)
      // unbounded queue is OK, since its naturally limited (threadPoolSize + # of transports (used for flushing))
      executor = new ThreadPoolExecutor (threadPoolSize, threadPoolSize,
                                         Long.MAX_VALUE, TimeUnit.NANOSECONDS,
                                         new LinkedBlockingQueue<> ());
      executor.prestartAllCoreThreads ();
   }

   /**
    * Promote a new leader.
    *
    * @param task task to execute by a new leader.
    */
   public void promoteLeader( Runnable task )
   {
      //System.err.println("[promoteLeader] by " + Thread.currentThread().getName());
      execute (task);
   }

   /**
    * Execute given task.
    *
    * @param task task to execute.
    */
   public void execute( Runnable task )
   {
      try
      {
         executor.execute (task);
      }
      catch ( Throwable th )
      {
         /* noop */
         logger.log( Level.SEVERE, "Unexpected exception caught in one of the LF thread-pool thread.", th);
      }
   }

   /**
    * Shutdown.
    */
   public void shutdown()
   {
      if ( shutdown.get() )
      {
         return;
      }
      shutdown.set( true );
      executor.shutdown ();
      try
      {
         // NOTE: if thead pool is shutdown from one of its threads, this will always block for 1s
         if ( !executor.awaitTermination (1, TimeUnit.SECONDS) )
         {
            executor.shutdownNow();
         }
      }
      catch ( InterruptedException ie )
      { /* noop */ }
   }

} public class CARepeaterServiceManager
{

/*- Public attributes --------------------------------------------------------*/
/*- Private attributes -------------------------------------------------------*/

   private static final Logger logger = LibraryLogManager.getLogger( CARepeaterServiceManager.class );

   /**
    * Map of service instances (= CA Repeaters running on particular port) and
    * the number of times the services has been requested or cancelled.
    */
   private final Map<CARepeaterServiceInstance, Integer> serviceInterestMap = new HashMap<>();

/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/

   /**
    * Handles a request for the CA Repeater Service on the specified port.
    * <p>
    * The request for CA Repeater services does not always result in an instance
    * of the CA Repeater from being started since various conditions may
    * prevent it. These situations are considered normal and are logged
    * but do not result in any exceptions being raised.
    * <p>
    * Things that are handled as part of the normal processing:
    * <ol>
    *    <li>The library configuration has disabled the service.</li>
    *    <li>There have been previous requests for the CA Repeater Service on this
    *    port which have not yet been cancelled.</li>
    *    <li>An instance of the CA Repeater was already started by someone else
    *    on the local machine.</li>
    *    <li>The attempt to start a new CA Repeater service instance results
    *    in an unexpected error.</li>
    * </ol>
    *
    * @param repeaterPort the port on which the service is requested.
    */
   public void requestServiceOnPort( int repeaterPort )
   {
      synchronized ( CARepeaterServiceManager.class )
      {
         logger.fine( "Processing request for CA Repeater Services on port '" + repeaterPort + "." );

         final int currentInterestLevel = getServiceInterestLevelForPort( repeaterPort );
         if ( getServiceInterestLevelForPort( repeaterPort ) == 0 )
         {
            logger.fine( "Starting the port " + repeaterPort + " CA Repeater Service instance." );
            final CARepeaterServiceInstance serviceInstance = new CARepeaterServiceInstance( repeaterPort );
            serviceInterestMap.put( serviceInstance, 1 );
            serviceInstance.activate();
         }
         else
         {
            logger.finer( "Increasing the interest level in the port " + repeaterPort  + " CA Repeater Service instance." );
            increaseServiceInterestLevelForPort( repeaterPort );
         }
      }
   }

   /**
    * Cancels request for the CA Repeater Service on the specified port.
    * <p>
    * The request for a CA Repeater does not always result in an instance
    * of the CA Repeater from being shutdown since various conditions may
    * prevent it. These situations are considered normal and are logged
    * but do not result in any exceptions being raised.
    * <p>
    * Things that are handled as part of the normal processing:
    * <ol>
    *    <li>The library configuration has disabled the service.</li>
    *    <li>There were no previous service requests for the CA Repeater on
    *    this port.</li>
    *    <li>There were no previous service requests for the CA Repeater on
    *    this port.</li>
    * </ol>
    *
    * @param repeaterPort the port on which the service is requested.
    */
   public void cancelServiceRequestOnPort( int repeaterPort )
   {
      synchronized ( CARepeaterServiceManager.class )
      {
         logger.fine( "Processing request for cancellation of CA Repeater Services on port: '" + repeaterPort + "'." );

         final int currentInterestLevel = getServiceInterestLevelForPort( repeaterPort );
         switch ( currentInterestLevel )
         {
            case 0:
               logger.warning( "The CA Repeater Service has received no previous service requests for port " + repeaterPort + "." );
               break;

            case 1:
               logger.fine( "Shutting down the port " + repeaterPort + " CA Repeater Service instance." );
               final Optional<CARepeaterServiceInstance> optInstance = getServiceInstanceFor( repeaterPort );
               Validate.validState( optInstance.isPresent() );
               serviceInterestMap.remove( optInstance.get() );
               optInstance.get().shutdown();
               break;

            default:
               logger.finer( "Reducing the interest level in the port " + repeaterPort  + " CA Repeater Service instance." );
               reduceServiceInterestlevelForPort( repeaterPort );
               break;
         }
      }
   }


/*- Package-level access methods ---------------------------------------------*/

   /**
    * This method is provided for test purposes only.
    *
    * @return the number of unique service instances that are currently active.
    */
   int getServiceInstances()
   {
      return serviceInterestMap.size();
   }

/*- Private methods ----------------------------------------------------------*/

   /**
    * Returns the level of interest in a CA Repeater service interest
    * (= instance of a CA Repeater running on the specified port).
    *
    * The interest level is incremented every time a new service
    * request is made and decremented every time it is cancelled.
    *
    * @param port the port of interest.
    * @return the result.
    */
   private int getServiceInterestLevelForPort( int port )
   {
      Validate.inclusiveBetween( 0, 65535, port );

      final Optional<Integer> currentInterestLevel = serviceInterestMap.entrySet().stream().filter( x -> x.getKey().getPort() == port ).map( Map.Entry::getValue ).findFirst();
      return currentInterestLevel.orElse( 0 );
   }

   /**
    * Returns the service instance, if any, associated with the specified port.
    *
    * @param port the port of interest.
    * @return the result.
    */
   private Optional<CARepeaterServiceInstance> getServiceInstanceFor( int port )
   {
      Validate.inclusiveBetween( 0, 65535, port );
      return serviceInterestMap.keySet().stream().filter( integer -> integer.getPort() == port ).findFirst();
   }

   /**
    * Increases the level of interest in the service instance associated with
    * the specified port.
    *
    * @param port the target port.
    *
    * @throws IllegalArgumentException if the port is out of range.
    * @throws IllegalStateException if the service instance on this port was never requested.
    * @throws IllegalStateException if the current interest level wasn't at least 1.
    */
   private void increaseServiceInterestLevelForPort( int port )
   {
      Validate.inclusiveBetween( 0, 65535, port );

      final Optional<CARepeaterServiceInstance> instance = getServiceInstanceFor( port );
      Validate.validState( instance.isPresent() );

      final int currentInterestLevel = getServiceInterestLevelForPort( port );
      Validate.validState( currentInterestLevel >= 1 );

      final int newInterestLevel = getServiceInterestLevelForPort( port ) + 1;
      serviceInterestMap.put( instance.get(), newInterestLevel );
   }

   /**
    * Reduces the level of interest in the service instance associated with
    * the specified port.
    *
    * @param port the target port.
    *
    * @throws IllegalArgumentException if the port is out of range.
    * @throws IllegalStateException if the service instance on this port was never requested.
    * @throws IllegalStateException if the current interest level wasn't at least 1.
    */

   private void reduceServiceInterestlevelForPort( int port )
   {
      Validate.inclusiveBetween( 0, 65535, port );

      final Optional<CARepeaterServiceInstance> instance = getServiceInstanceFor( port );
      Validate.validState( instance.isPresent() );

      final int currentInterestLevel = getServiceInterestLevelForPort( port );
      Validate.validState( currentInterestLevel >= 1 );

      final int newInterestLevel = currentInterestLevel - 1;
      serviceInterestMap.put( instance.get(), newInterestLevel );
   }

/*- Nested Classes -----------------------------------------------------------*/

} public class ChannelSearchManager
{

/*- Public attributes --------------------------------------------------------*/
/*- Private attributes -------------------------------------------------------*/

   private static final int MIN_SEND_INTERVAL_MS_DEFAULT = 100;
   private static final int MAX_SEND_INTERVAL_MS_DEFAULT = 30_000;
   private static final int INTERVAL_MULTIPLIER_DEFAULT = 2;

   private static final int MESSAGE_COALESCENCE_TIME_MS = 3;

   private static final int MAX_NUMBER_IMMEDIATE_PACKETS = 5;
   private static final int IMMEDIATE_PACKETS_DELAY_MS = 10;

   private static final Logger logger = LibraryLogManager.getLogger( ChannelSearchManager.class );

   /**
    * Minimum time between sending packets (ms).
    */
   private final long minSendInterval;

   /**
    * Maximum time between sending packets (ms).
    */
   private final long maxSendInterval;

   /**
    * Search interval multiplier, i.e. multiplier for exponential back-off.
    */
   private final int intervalMultiplier;

   private final SearchTimer timer = new SearchTimer();
   private final AtomicBoolean canceled = new AtomicBoolean();

   private final AtomicInteger immediatePacketCount = new AtomicInteger();
   private final AtomicInteger channelCount = new AtomicInteger();

   /**
    * Broadcast transport.
    */
   private final UdpBroadcastTransport udpBroadcastTransport;

   /**
    * Search (datagram) sequence number.
    */
   private final AtomicInteger sequenceNumber = new AtomicInteger( 0 );

   /**
    * Send byte buffer (frame)
    */
   private final ByteBuffer sendBuffer;


/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/

   /**
    * Constructor.
    *
    * @param udpBroadcastTransport transport
    */
   public ChannelSearchManager( UdpBroadcastTransport udpBroadcastTransport )
   {
      this.udpBroadcastTransport = udpBroadcastTransport;

      minSendInterval = MIN_SEND_INTERVAL_MS_DEFAULT;
      maxSendInterval = MAX_SEND_INTERVAL_MS_DEFAULT;
      intervalMultiplier = INTERVAL_MULTIPLIER_DEFAULT;

      // create and initialize send buffer
      sendBuffer = ByteBuffer.allocateDirect( Constants.MAX_UDP_SEND );
      initializeSendBuffer ();
   }


/*- Public methods -----------------------------------------------------------*/

   /**
    * Register channel.
    *
    * @param channel the channel to register.
    * @return true if the channel was successfully registered.
    */
   public boolean registerChannel( ChannelImpl<?> channel )
   {
      if ( canceled.get() )
      {
         return false;
      }

      final ChannelSearchTimerTask timerTask = new ChannelSearchTimerTask( channel );
      channel.setTimerId( timerTask );

      timer.executeAfterDelay( MESSAGE_COALESCENCE_TIME_MS, timerTask );
      channelCount.incrementAndGet();
      return true;
   }

   /**
    * Unregister channel.
    *
    * @param channel channel to unregister
    */
   public void unregisterChannel( ChannelImpl<?> channel )
   {
      if ( canceled.get() )
      {
         return;
      }

      final Object timerTask = channel.getTimerId();
      if ( timerTask != null )
      {
         SearchTimer.cancel( timerTask );
         channel.setTimerId( null );
      }
      channelCount.decrementAndGet();
   }

   /**
    * Get number of registered channels.
    *
    * @return number of registered channels.
    */
   public int registeredChannelCount()
   {
      return channelCount.get();
   }

   /**
    * Beacon anomaly detected.
    * Boost searching of all channels.
    */
   public void beaconAnomalyNotify()
   {
      logger.fine( "A beacon anomaly has been detected." );
      if ( canceled.get() )
      {
         return;
      }
      logger.fine( "Reinstigating channel search." );
      timer.rescheduleAllAfterDelay(0 );
   }

   /**
    * Cancel.
    */
   public void cancel()
   {
      if ( canceled.getAndSet(true) )
      {
         return;
      }
      timer.shutDown();
   }


/*- Private methods ----------------------------------------------------------*/

   /**
    * Initialize send buffer.
    */
   private void initializeSendBuffer()
   {
      sendBuffer.clear ();

      // put version message
      sequenceNumber.incrementAndGet();
      Messages.generateVersionRequestMessage( udpBroadcastTransport, sendBuffer, (short) 0, sequenceNumber.get(), true );
   }

   /**
    * Flush send buffer.
    */
   private synchronized void flushSendBuffer()
   {
      if ( immediatePacketCount.incrementAndGet() >= MAX_NUMBER_IMMEDIATE_PACKETS )
      {
         try
         {
            Thread.sleep( IMMEDIATE_PACKETS_DELAY_MS );
         }
         catch ( InterruptedException ex )
         {
            // noop
         }
         immediatePacketCount.set( 0 );
      }

      udpBroadcastTransport.send( sendBuffer );
      initializeSendBuffer();
   }

   /**
    * Generate (put on send buffer) search request
    *
    * @param channel       channel
    * @param allowNewFrame flag indicating if new search request message is allowed to be put in new frame.
    * @return <code>true</code> if new frame was sent.
    */
   private synchronized boolean generateSearchRequestMessage( ChannelImpl<?> channel, @SuppressWarnings( "SameParameterValue" ) boolean allowNewFrame )
   {
      boolean success = channel.generateSearchRequestMessage (udpBroadcastTransport, sendBuffer);
      // buffer full, flush
      if ( !success )
      {
         flushSendBuffer();
         if ( allowNewFrame )
         {
            channel.generateSearchRequestMessage( udpBroadcastTransport, sendBuffer );
         }
         return true;
      }

      return false;
   }

   /**
    * Search response received notification.
    *
    * @param channel found channel.
    */
   public void searchResponse( ChannelImpl<?> channel )
   {
      unregisterChannel (channel);
   }


/*- Nested Classes -----------------------------------------------------------*/

   private class ChannelSearchTimerTask extends SearchTimer.TimerTask
   {
      private final ChannelImpl<?> channel;

      ChannelSearchTimerTask( ChannelImpl<?> channel )
      {
         this.channel = channel;
      }

      public long timeout()
      {
         // send search message
         generateSearchRequestMessage( channel, true );

         if ( !timer.hasNext( MESSAGE_COALESCENCE_TIME_MS ) )
         {
            flushSendBuffer();
            immediatePacketCount.set( 0 );
         }

         // reschedule
         long dT = getDelay();

         dT *= intervalMultiplier;
         if ( dT > maxSendInterval )
         {
            dT = maxSendInterval;
         }
         if ( dT < minSendInterval )
         {
            dT = minSendInterval;
         }

         return dT;
      }
   }

} @SuppressWarnings( "UnnecessaryLocalVariable" )
public class IntHashMap<T>
{

   /**
    * The hash table data.
    */
   private transient Entry[] table;

   /**
    * The total number of entries in the hash table.
    */
   private transient int count;

   /**
    * The table is rehashed when its size exceeds this threshold.  (The
    * value of this field is (int)(capacity * loadFactor).)
    *
    * @serial
    */
   private int threshold;

   /**
    * The load factor for the hashtable.
    *
    * @serial
    */
   private final float loadFactor;

   /**
    * Entry object pool.
    */
   private final transient EntryObjectPool pool;

   /**
    * <p>Inner class that acts as a data structure to create a new entry in the
    * table.</p>
    */
   private static class Entry
   {
      int hash;
      //int key;
      Object value;
      Entry next;

      /**
       * <p>Create a new entry with the given values.</p>
       *
       * @param hash  The code used to hash the object with
       * @param key   The key used to enter this in the table
       * @param value The value for this key
       * @param next  A reference to the next entry in the table
       */
      protected Entry( int hash, int key, Object value, Entry next )
      {
         initialize (hash, key, value, next);
      }

      /**
       * <p>Set given values.</p>
       *
       * @param hash  The code used to hash the object with
       * @param key   The key used to enter this in the table
       * @param value The value for this key
       * @param next  A reference to the next entry in the table
       */
      public void initialize( int hash, int key, Object value, Entry next )
      {
         this.hash = hash;
         //this.key = key;
         this.value = value;
         this.next = next;
      }

   }

   /**
    * <p>Object pool implementation for Entry class.</p>
    */
   private static class EntryObjectPool
   {

      /**
       * Pool of reusable objects.
       */
      private final ArrayList<Entry> pool;
      private int lastPos = -1;

      /**
       * Constructor.
       *
       * @param initialCapacity initial capacity of the pool.
       */
      public EntryObjectPool( int initialCapacity )
      {
         pool = new ArrayList<> (initialCapacity);
      }

      /**
       * Get <code>Entry</code> object from the object pool.
       *
       * @param hash  The code used to hash the object with
       * @param key   The key used to enter this in the table
       * @param value The value for this key
       * @param next  A reference to the next entry in the table
       * @return   <code>Entry</code> instance.
       */
      public Entry getEntry( int hash, int key, Object value, Entry next )
      {
         if ( lastPos == -1 )
         {
            return new Entry (hash, key, value, next);
         }
         else
         {
            Entry entry = pool.remove (lastPos--);
            entry.initialize (hash, key, value, next);
            return entry;
         }
      }

      /**
       * Put (return) <code>Entry</code> object to the object pool.
       *
       * @param entry entry to put.
       */
      public void putEntry( Entry entry )
      {
         // NOTE: be sure that entry.value and entry.next are null
         pool.add (entry);
         lastPos++;
      }
   }

   /**
    * <p>Constructs a new, empty hashtable with a default capacity and load
    * factor, which is <code>20</code> and <code>0.75</code> respectively.</p>
    */
   public IntHashMap()
   {
      this (20, 0.75f);
   }

   /**
    * <p>Constructs a new, empty hashtable with the specified initial capacity
    * and default load factor, which is <code>0.75</code>.</p>
    *
    * @param initialCapacity the initial capacity of the hashtable.
    * @throws IllegalArgumentException if the initial capacity is less
    *                                  than zero.
    */
   public IntHashMap( int initialCapacity )
   {
      this (initialCapacity, 0.75f);
   }

   /**
    * <p>Constructs a new, empty hashtable with the specified initial
    * capacity and the specified load factor.</p>
    *
    * @param initialCapacity the initial capacity of the hashtable.
    * @param loadFactor      the load factor of the hashtable.
    * @throws IllegalArgumentException if the initial capacity is less
    *                                  than zero, or if the load factor is nonpositive.
    */
   public IntHashMap( int initialCapacity, float loadFactor )
   {
      super ();
      if ( initialCapacity < 0 )
      {
         throw new IllegalArgumentException ("Illegal Capacity: " + initialCapacity);
      }
      if ( loadFactor <= 0 )
      {
         throw new IllegalArgumentException ("Illegal Load: " + loadFactor);
      }
      if ( initialCapacity == 0 )
      {
         initialCapacity = 1;
      }

      this.loadFactor = loadFactor;
      table = new Entry[ initialCapacity ];
      threshold = (int) (initialCapacity * loadFactor);

      pool = new EntryObjectPool (initialCapacity);
   }

   /**
    * <p>Returns the number of keys in this hashtable.</p>
    *
    * @return the number of keys in this hashtable.
    */
   public int size()
   {
      return count;
   }

   /**
    * <p>Tests if this hashtable maps no keys to values.</p>
    *
    * @return <code>true</code> if this hashtable maps no keys to values;
    * <code>false</code> otherwise.
    */
   public boolean isEmpty()
   {
      return count == 0;
   }

   /**
    * <p>Tests if some key maps into the specified value in this hashtable.
    * This operation is more expensive than the <code>containsKey</code>
    * method.</p>
    *
    * <p>Note that this method is identical in functionality to containsValue,
    * (which is part of the Map interface in the collections framework).</p>
    *
    * @param value a value to search for.
    * @return <code>true</code> if and only if some key maps to the
    * <code>value</code> argument in this hashtable as
    * determined by the <code>equals</code> method;
    * <code>false</code> otherwise.
    * @throws NullPointerException if the value is <code>null</code>.
    * @see #containsKey(int)
    * @see #containsValue(Object)
    * @see java.util.Map
    */
   public boolean contains( T value )
   {
      if ( value == null )
      {
         throw new NullPointerException ();
      }

      final Entry[] tab = table;
      for ( int i = tab.length; i-- > 0; )
      {
         for ( Entry e = tab[ i ]; e != null; e = e.next )
         {
            if ( e.value.equals (value) )
            {
               return true;
            }
         }
      }
      return false;
   }

   /**
    * <p>Returns <code>true</code> if this HashMap maps one or more keys
    * to this value.</p>
    *
    * <p>Note that this method is identical in functionality to contains
    * (which predates the Map interface).</p>
    *
    * @param value value whose presence in this HashMap is to be tested.
    * @return boolean <code>true</code> if the value is contained
    * @see java.util.Map
    * @since JDK1.2
    */
   public boolean containsValue( T value )
   {
      return contains (value);
   }

   /**
    * <p>Tests if the specified object is a key in this hashtable.</p>
    *
    * @param key possible key.
    * @return <code>true</code> if and only if the specified object is a
    * key in this hashtable, as determined by the <code>equals</code>
    * method; <code>false</code> otherwise.
    * @see #contains(Object)
    */
   public boolean containsKey( int key )
   {
      final Entry[] tab = table;
      final int hash = key;
      final int index = (hash & 0x7FFFFFFF) % tab.length;
      for ( Entry e = tab[ index ]; e != null; e = e.next )
      {
         if ( e.hash == hash )
         {
            return true;
         }
      }
      return false;
   }

   /**
    * <p>Returns the value to which the specified key is mapped in this map.</p>
    *
    * @param key a key in the hashtable.
    * @return the value to which the key is mapped in this hashtable;
    * <code>null</code> if the key is not mapped to any value in
    * this hashtable.
    * @see #put(int, Object)
    */
   @SuppressWarnings( "unchecked" )
   public T get( int key )
   {
      final Entry[] tab = table;
      final int hash = key;
      final int index = (hash & 0x7FFFFFFF) % tab.length;
      for ( Entry e = tab[ index ]; e != null; e = e.next )
      {
         if ( e.hash == hash )
         {
            return (T) e.value;
         }
      }
      return null;
   }

   /**
    * <p>Increases the capacity of and internally reorganizes this
    * hashtable, in order to accommodate and access its entries more
    * efficiently.</p>
    *
    * <p>This method is called automatically when the number of keys
    * in the hashtable exceeds this hashtable's capacity and load
    * factor.</p>
    */
   protected void rehash()
   {
      final int oldCapacity = table.length;
      final Entry[] oldMap = table;

      final int newCapacity = oldCapacity * 2 + 1;
      final Entry[] newMap = new Entry[ newCapacity ];

      threshold = (int) (newCapacity * loadFactor);
      table = newMap;

      for ( int i = oldCapacity; i-- > 0; )
      {
         for ( Entry old = oldMap[ i ]; old != null; )
         {
            Entry e = old;
            old = old.next;

            int index = (e.hash & 0x7FFFFFFF) % newCapacity;
            e.next = newMap[ index ];
            newMap[ index ] = e;
         }
      }
   }

   /**
    * <p>Maps the specified <code>key</code> to the specified
    * <code>value</code> in this hashtable. The key cannot be
    * <code>null</code>. </p>
    *
    * <p>The value can be retrieved by calling the <code>get</code> method
    * with a key that is equal to the original key.</p>
    *
    * @param key   the hashtable key.
    * @param value the value.
    * @return the previous value of the specified key in this hashtable,
    * or <code>null</code> if it did not have one.
    * @throws NullPointerException if the key is <code>null</code>.
    * @see #get(int)
    */
   public Object put( int key, T value )
   {
      // Makes sure the key is not already in the hashtable.
      Entry[] tab = table;
      int hash = key;
      int index = (hash & 0x7FFFFFFF) % tab.length;
      for ( Entry e = tab[ index ]; e != null; e = e.next )
      {
         if ( e.hash == hash )
         {
            Object old = e.value;
            e.value = value;
            return old;
         }
      }

      if ( count >= threshold )
      {
         // Rehash the table if the threshold is exceeded
         rehash ();

         tab = table;
         index = (hash & 0x7FFFFFFF) % tab.length;
      }

      // Creates the new entry.
      //Entry e = new Entry(hash, key, value, tab[index]);
      Entry e = pool.getEntry (hash, key, value, tab[ index ]);
      tab[ index ] = e;
      count++;
      return null;
   }

   /**
    * <p>Removes the key (and its corresponding value) from this
    * hashtable.</p>
    *
    * <p>This method does nothing if the key is not present in the
    * hashtable.</p>
    *
    * @param key the key that needs to be removed.
    * @return the value to which the key had been mapped in this hashtable,
    * or <code>null</code> if the key did not have a mapping.
    */
   @SuppressWarnings( "unchecked" )
   public T remove( int key )
   {
      Entry[] tab = table;
      int hash = key;
      int index = (hash & 0x7FFFFFFF) % tab.length;
      for ( Entry e = tab[ index ], prev = null; e != null; prev = e, e = e.next )
      {
         if ( e.hash == hash )
         {
            if ( prev != null )
            {
               prev.next = e.next;
            }
            else
            {
               tab[ index ] = e.next;
            }
            count--;
            Object oldValue = e.value;
            e.value = null;

            e.next = null;
            pool.putEntry (e);

            return (T) oldValue;
         }
      }
      return null;
   }

   /**
    * <p>Clears this hashtable so that it contains no keys.</p>
    */
   public void clear()
   {
      final Entry[] tab = table;
      for ( int index = tab.length; --index >= 0; )
      {
         final Entry e = tab[ index ];
         if ( e != null )
         {
            e.value = null;
            e.next = null;
            pool.putEntry (e);
         }
         tab[ index ] = null;
      }
      count = 0;
   }

   /**
    * <p>Copies values to array. Note that array capacity has to be large enough.</p>
    *
    * @param arr array to be filled, new instance returned if given not large enough or <code>null</code>.
    * @return array of values.
    */
   @SuppressWarnings( "unchecked" )
   public T[] toArray( T[] arr )
   {
      if ( arr == null || arr.length < count )
      {
         // TODO: Note: March 2020: this looks like a bug to me. But without a unit
         //  test to confirm  this I will not change things.
         arr = (T[]) Array.newInstance( arr.getClass().getComponentType(), count );
      }

      int pos = 0;
      final Entry[] tab = table;
      for ( int i = tab.length; i-- > 0; )
      {
         for ( Entry e = tab[ i ]; e != null; e = e.next )
         {
            arr[ pos++ ] = (T) e.value;
         }
      }
      return arr;
   }

} public class LibraryLogManager
{

/*- Public attributes --------------------------------------------------------*/
/*- Private attributes -------------------------------------------------------*/

   private static final StreamHandler flushingStandardOutputStreamHandler;

   static
   {
      // Could consider here setting the default locale for this instance of
      // the Java Virtual Machine. But currently (2020-05-10) it is not
      // considered that the library "owns" this decision.
      // Locale.setDefault( Locale.ROOT );
      // Note: the definition below determines the format of all log messages
      // emitted by the CA library.
      System.setProperty( "java.util.logging.SimpleFormatter.format", "%1$tF %1$tT.%1$tL %3$s %4$s %5$s %6$s %n" );

      // Create a stream handler that is like the normal output stream handler except
      // it will ensure that each message gets flushed to the console. Note: this may
      // impose some performance penalty during testing if copious messages are
      // being logged. During normal CA usage the library does not emit many
      // messages so in the normal situation this limitation should not be important.
      flushingStandardOutputStreamHandler = new StreamHandler( System.out, new SimpleFormatter() )
      {
         @Override
         public synchronized void publish( final LogRecord record )
         {
            super.publish( record );
            flush();
         }
      };

      // The output handler will emit all messages that are sent to it. It is up
      // to the individual loggers to determine what is appropriate for their
      // particular operating contexts.
      flushingStandardOutputStreamHandler.setLevel( Level.ALL );
   }


/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/

   /**
    * Returns a logger for the specified class that will send messages to the
    * standard output stream providing their log level exceeds the level defined 
    * by the CA_LIBRARY_LOG_LEVEL system property.
    *
    * When CA_LIBRARY_LOG_LEVEL is not defined all messages of Level.INFO and
    * above will be logged.
    *
    * @param clazz the class that the log messages will be associated with.
    * @return the configured logger.
    * 
    * throws NullPointerException if clazz was null.
    * throws IllegalArgumentException if the string token associated with
    * CA_LIBRARY_LOG_LEVEL could not be interpreted as a valid log level.
    */
   public static Logger getLogger( Class<?> clazz )
   {
      Validate.notNull( clazz );
      return getLogger( clazz, LibraryConfiguration.getInstance().getLibraryLogLevel() );
   }

   /**
    * Returns a logger for the specified class that will send messages to the
    * standard output stream providing their log level exceeds the specified
    * log level.
    *
    * @param clazz the class that the logger will be associated with
    *     when logging messages.
    *
    * @param logLevel Set the log level specifying which message levels
    *    will be sent to the standard output stream by this logger. Message levels
    *    lower than this value will be discarded. The value Level.OFF can be
    *    used to completely turn off logging.
    *
    * @return the configured logger.
    */
   public static Logger getLogger( Class<?> clazz, Level logLevel )
   {
      // Currently (2020-05-14) the simple name is used in the log. This
      // is considerably shorter than the fully qualified class name. Other
      // implementations are possible. For example the Spring framework
      // takes the approach of shortening the package names in the FQN to
      // a single character. Another approach would be to extract the name
      // automatically from the calling stack.
      final Logger logger = Logger.getLogger( clazz.getSimpleName() );
      logger.setUseParentHandlers( false );

      if ( logger.getHandlers().length == 0 )
      {
         logger.addHandler( flushingStandardOutputStreamHandler );
      }
      else
      {
         System.out.println( "\nWARNING: More than one logger defined for class: '" + clazz.getSimpleName() + "'.\n" );
      }
      logger.setLevel( logLevel );

      return logger;
   }


/*- Package-level methods ----------------------------------------------------*/

   /**
    * Provided to enable tests only.
    * @param logger the logger whose handlers are to be disposed.
    */
   static void disposeLogger( Logger logger)
   {
      if ( logger.getHandlers().length == 1 )
      {
         logger.removeHandler( flushingStandardOutputStreamHandler );
      }
   }

/*- Private methods ----------------------------------------------------------*/
/*- Nested Classes -----------------------------------------------------------*/

} public class InetAddressUtil
{

   /**
    * Get broadcast addresses.
    *
    * @param port port to be added to get socket address.
    * @return array of broadcast addresses with given port.
    */
   public static InetSocketAddress[] getBroadcastAddresses( int port )
   {
      Enumeration<NetworkInterface> nets;
      try
      {
         nets = NetworkInterface.getNetworkInterfaces();
      }
      catch ( SocketException se )
      {
         // fallback
         return new InetSocketAddress[] { new InetSocketAddress ("255.255.255.255", port) };
      }

      ArrayList<InetSocketAddress> list = new ArrayList<> (10);

      while ( nets.hasMoreElements () )
      {
         NetworkInterface net = nets.nextElement ();
         try
         {
            if ( net.isUp () )
            {
               final List<InterfaceAddress> interfaceAddresses = net.getInterfaceAddresses();
               if ( ! interfaceAddresses.isEmpty() )
               {
                  for ( InterfaceAddress addr : interfaceAddresses )
                  {
                     if ( addr.getBroadcast() != null )
                     {
                        InetSocketAddress isa = new InetSocketAddress(addr.getBroadcast(), port);
                        if ( !list.contains( isa ) )
                        {
                           list.add(isa);
                        }
                     }
                  }
               }
            }
         }
         catch ( Throwable th )
         {
            // some methods throw exceptions, some return null (and they shouldn't)
            // noop, skip that interface
         }
      }

      // fallback to loop
      if ( list.size () == 0 )
         list.add (new InetSocketAddress (InetAddress.getLoopbackAddress (), port));

      InetSocketAddress[] retVal = new InetSocketAddress[ list.size () ];
      list.toArray (retVal);
      return retVal;
   }

   /**
    * Convert an integer into an IPv4 INET address.
    *
    * @param addr integer representation of a given address.
    * @return IPv4 INET address.
    */
   public static InetAddress intToIPv4Address( int addr )
   {
      byte[] a = new byte[ 4 ];

      a[ 0 ] = (byte) ((addr >> 24) & 0xFF);
      a[ 1 ] = (byte) ((addr >> 16) & 0xFF);
      a[ 2 ] = (byte) ((addr >> 8) & 0xFF);
      a[ 3 ] = (byte) ((addr & 0xFF));

      InetAddress res = null;
      try
      {
         res = InetAddress.getByAddress (a);
      }
      catch ( UnknownHostException e )
      { /* noop */ }

      return res;
   }

   /**
    * Convert an IPv4 INET address to an integer.
    *
    * @param addr IPv4 INET address.
    * @return integer representation of a given address.
    * @throws IllegalArgumentException if the address is really an IPv6 address
    */
   public static int ipv4AddressToInt( InetAddress addr )
   {

      if ( addr instanceof Inet6Address )
         throw new IllegalArgumentException ("IPv6 address used in IPv4 context");

      byte[] a = addr.getAddress ();

      @SuppressWarnings( "UnnecessaryLocalVariable" )
      final int res = ((a[ 0 ] & 0xFF) << 24)
         | ((a[ 1 ] & 0xFF) << 16)
         | ((a[ 2 ] & 0xFF) << 8)
         | (a[ 3 ] & 0xFF);

      return res;
   }


   /**
    * Parse space delimited addresss[:port] string and return array of <code>InetSocketAddress</code>.
    *
    * @param list        space delimited addresss[:port] string.
    * @param defaultPort port take if not specified.
    * @return array of <code>InetSocketAddress</code>.
    */
   public static InetSocketAddress[] getSocketAddressList( String list, int defaultPort )
   {
      return getSocketAddressList (list, defaultPort, null);
   }

   /**
    * Parse space delimited address[:port] string and return array of <code>InetSocketAddress</code>.
    *
    * @param list        space delimited address[:port] string.
    * @param defaultPort port take if not specified.
    * @param appendList  list to be appended.
    * @return array of <code>InetSocketAddress</code>.
    */
   public static InetSocketAddress[] getSocketAddressList( String list, int defaultPort, InetSocketAddress[] appendList )
   {
      final ArrayList<InetSocketAddress> al = new ArrayList<> ();

      // parse string
      StringTokenizer st = new StringTokenizer (list);
      while ( st.hasMoreTokens () )
      {
         int port = defaultPort;
         String address = st.nextToken ();

         // check port
         int pos = address.indexOf (':');
         if ( pos >= 0 )
         {
            try
            {
               port = Integer.parseInt (address.substring (pos + 1));
            }
            catch ( NumberFormatException nfe )
            { /* noop */ }

            address = address.substring (0, pos);
         }

         // add parsed address
         al.add (new InetSocketAddress (address, port));
      }

      // copy to array
      int appendSize = (appendList == null) ? 0 : appendList.length;
      InetSocketAddress[] isar = new InetSocketAddress[ al.size () + appendSize ];
      al.toArray (isar);
      if ( appendSize > 0 )
      {
         System.arraycopy(appendList, 0, isar, al.size(), appendSize);
      }
      return isar;
   }

   private static String hostName = null;

   public static synchronized String getHostName()
   {
      if ( hostName == null )
      {
         // default fallback
         hostName = "localhost";

         try
         {
            InetAddress localAddress = InetAddress.getLocalHost ();
            hostName = localAddress.getHostName ();
         }
         catch ( Throwable uhe )
         {   // not only UnknownHostException
            // try with environment variable
            try
            {
               String envHN = System.getenv (Constants.CA_HOSTNAME_KEY);
               if ( envHN != null )
                  hostName = envHN;
            }
            catch ( Throwable th )
            {
               // in case not supported by JVM/OS
            }

            // and system property (overrides env. var.)
            hostName = System.getProperty (Constants.CA_HOSTNAME_KEY, hostName);
         }

         if ( System.getProperties ().contains (Constants.CA_STRIP_HOSTNAME) )
         {
            int dotPos = hostName.indexOf ('.');
            if ( dotPos > 0 )
               hostName = hostName.substring (0, dotPos);
         }
      }

      return hostName;
   }

} public final class NamedLockPattern
{

   /**
    * Map of (named) locks.
    */
   private final Map<Object, ReferenceCountingLock> namedLocks = new HashMap<> ();

   /**
    * Acquire synchronization lock for named object.
    *
    * @param   name   name of the object whose lock to acquire.
    * @param   msec   the number of milliseconds to wait.
    * An argument less than or equal to zero means not to wait at all.
    * @return   <code>true</code> if acquired, <code>false</code> otherwise.
    */
   public boolean acquireSynchronizationObject( Object name, long msec )
   {
      ReferenceCountingLock lock;

      synchronized ( namedLocks )
      {
         // get synchronization object
         lock = namedLocks.get (name);

         // none is found, create and return new one
         // increment references
         if ( lock == null )
         {
            lock = new ReferenceCountingLock ();
            namedLocks.put (name, lock);
         }
         else
            lock.increment ();
      }

      boolean success = lock.acquire (msec);

      if ( !success )
         releaseSynchronizationObject (name, false);

      return success;
   }

   /**
    * Release synchronization lock for named object.
    *
    * @param   name   name of the object whose lock to release.
    */
   public void releaseSynchronizationObject( Object name )
   {
      releaseSynchronizationObject (name, true);
   }

   /**
    * Release synchronization lock for named object.
    *
    * @param   name   name of the object whose lock to release.
    * @param   release   set to <code>false</code> if there is no need to call release
    * on synchronization lock.
    */
   public void releaseSynchronizationObject( Object name, boolean release )
   {
      synchronized ( namedLocks )
      {
         // get synchronization object
         ReferenceCountingLock lock = namedLocks.get (name);

         // release lock
         if ( lock != null )
         {
            // if there only one current lock exists
            // remove it from the map
            if ( lock.decrement () <= 0 )
               namedLocks.remove (name);

            // release the lock
            if ( release )
               lock.release ();
         }
      }
   }

} public class Constants
{

/*- Public attributes --------------------------------------------------------*/

   public enum ChannelProperties
   {
      nativeType, nativeTypeCode, remoteAddress, nativeElementCount
   }

   /**
    * String value of the JVM property key which specifies whether to strip the hostname returned by
    * InetAddress.getLocalHost().getHostName().
    */
   public static final String CA_STRIP_HOSTNAME = "CA_STRIP_HOSTNAME";

   /**
    * String value of the JVM property key to provide (override) hostname.
    */
   public static final String CA_HOSTNAME_KEY = "HOSTNAME";

   /**
    * Minimal priority.
    */
   public static final short CHANNEL_PRIORITY_MIN = 0;

   /**
    * Maximal priority.
    */
   public static final short CHANNEL_PRIORITY_MAX = 99;

   /**
    * Default priority.
    */
   public static final short CHANNEL_PRIORITY_DEFAULT = CHANNEL_PRIORITY_MIN;

   /**
    * DB links priority.
    */
   public static final short CHANNEL_PRIORITY_LINKS_DB = CHANNEL_PRIORITY_MAX;

   /**
    * Archive priority.
    */
   public static final short CHANNEL_PRIORITY_ARCHIVE = (short) ((CHANNEL_PRIORITY_MAX + CHANNEL_PRIORITY_MIN) / 2);

   /**
    * OPI priority.
    */
   public static final short CHANNEL_PRIORITY_OPI = CHANNEL_PRIORITY_MIN;

   /* -------- Core CA constants -------- */

   /**
    * CA protocol major revision (implemented by this library).
    */
   public static final short CA_MAJOR_PROTOCOL_REVISION = 4;

   /**
    * CA protocol minor revision (implemented by this library).
    */
   public static final short CA_MINOR_PROTOCOL_REVISION = 13;

   /**
    * Unknown CA protocol minor revision.
    */
   public static final short CA_UNKNOWN_MINOR_PROTOCOL_REVISION = 0;

   /**
    * Initial delay in milliseconds between creation of a new Context and the
    * first attempt to register with the CA Repeater.
    */
   public static final int CA_REPEATER_INITIAL_REGISTRATION_DELAY = 500;

   /**
    * CA Repeater attempted registration interval in milliseconds.
    */
   public static final int CA_REPEATER_REGISTRATION_INTERVAL = 60_000;

   /**
    * CA protocol message header size.
    */
   public static final short CA_MESSAGE_HEADER_SIZE = 16;

   /**
    * CA protocol message extended header size.
    */
   public static final short CA_EXTENDED_MESSAGE_HEADER_SIZE = (short) (CA_MESSAGE_HEADER_SIZE + 8);

   /**
    * UDP maximum send message size.
    * MAX_UDP: 1500 (max of Ethernet and 802.{2,3} MTU) - 20(IP) - 8(UDP)
    * (the MTU of Ethernet is currently independent of its speed variant)
    */
   public static final int MAX_UDP_SEND = 1024;

   /**
    * UDP maximum receive message size.
    */
   public static final int MAX_UDP_RECV = 0xFFFF + 16;

   /**
    * TCP maximum receive message size.
    */
   public static final int MAX_TCP_RECV = 1024 * 16 + CA_EXTENDED_MESSAGE_HEADER_SIZE;

   /**
    * Default priority (corresponds to POSIX SCHED_OTHER)
    */
   public static final short CA_DEFAULT_PRIORITY = 0;

   /**
    * Read access right mask.
    */
   public static final int CA_PROTO_ACCESS_RIGHT_READ = 1;

   /**
    * Write access right mask.
    */
   public static final int CA_PROTO_ACCESS_RIGHT_WRITE = 1 << 1;

   /**
    * Do not require response for CA search request.
    */
   public static final short CA_SEARCH_DONTREPLY = 5;

   /**
    * Require response (even if not found) for CA search request over TCP.
    */
   public static final short CA_SEARCH_DOREPLY = 10;

   /**
    * Echo (state-of-health message) response timeout in ms.
    */
   public static final long CA_ECHO_TIMEOUT = 5000;

   /**
    * Max. (requested) string size.
    */
   public static final int MAX_STRING_SIZE = 40;

   /**
    * Unreasonable channel name length.
    */
   public static final int UNREASONABLE_CHANNEL_NAME_LENGTH = 500;

/*- Private attributes -------------------------------------------------------*/
/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/
/*- Private methods ----------------------------------------------------------*/
/*- Public methods -----------------------------------------------------------*/
/*- Private methods ----------------------------------------------------------*/
/*- Nested Classes -----------------------------------------------------------*/

}
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.lang3.Validate;
import org.epics.ca.Channel;
import org.epics.ca.Constants;
import org.epics.ca.impl.monitor.MonitorNotificationServiceFactoryCreator;
import org.epics.ca.impl.monitor.MonitorNotificationServiceFactory;
import org.epics.ca.impl.reactor.Reactor;
import org.epics.ca.impl.reactor.ReactorHandler;
import org.epics.ca.impl.reactor.lf.LeaderFollowersHandler;
import org.epics.ca.impl.reactor.lf.LeaderFollowersThreadPool;
import org.epics.ca.impl.repeater.CARepeaterServiceManager;
import org.epics.ca.impl.search.ChannelSearchManager;
import org.epics.ca.util.IntHashMap;
import org.epics.ca.util.logging.LibraryLogManager;
import org.epics.ca.util.net.InetAddressUtil;
import org.epics.ca.util.sync.NamedLockPattern;
import static org.epics.ca.Constants.*;
/*- Package Declaration ------------------------------------------------------*/

package org.epics.ca.impl;

/*- Imported packages --------------------------------------------------------*/





/*- Interface Declaration ----------------------------------------------------*/
/*- Class Declaration --------------------------------------------------------*/

public class ContextImpl implements AutoCloseable
{

/*- Public attributes --------------------------------------------------------*/
/*- Private static attributes ------------------------------------------------*/

   // The CA protocol only works on IPV4 enabled network stacks.
   // The definition below ensures that the external users of the library do not
   // have to explicitly set these definitions when using the library.
   static {
      System.setProperty( "java.net.preferIPv4Stack", "true" );
      System.setProperty( "java.net.preferIPv6Stack", "false" );
   }

   private static final Logger logger = LibraryLogManager.getLogger( ContextImpl.class );

   /**
    * Context instance.
    */
   private static final int LOCK_TIMEOUT = 20 * 1000;   // 20s

   /**
    * CA Repeater Service manager.
    */
   private static final CARepeaterServiceManager caRepeaterServiceManager = new CARepeaterServiceManager();

/*- Private final attributes (initialised immediately) -----------------------*/

   /**
    * Context instance.
    */
   private final NamedLockPattern namedLocker = new NamedLockPattern();

   /**
    * Timer.
    */
   private final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor();

   /**
    * General executor service (e.g. event dispatcher).
    */
   private final ExecutorService executorService = Executors.newSingleThreadExecutor();

   /**
    * TCP transport registry.
    */
   private final TransportRegistry transportRegistry = new TransportRegistry();

   /**
    * UDP broadcast transport - used for channel search requests and registration
    * with the CA Repeater.
    */
   private final AtomicReference<UdpBroadcastTransport> udpBroadcastTransportRef = new AtomicReference<>();

   /**
    * Beacon handler map.
    */
   private final Map<InetSocketAddress, BeaconHandler> beaconHandlers = new HashMap<>();

   /**
    * Map of channels (keys are CIDs).
    */
   private final IntHashMap<ChannelImpl<?>> channelsByCID = new IntHashMap<>();

   /**
    * Map of requests (keys are IOID).
    */
   private final IntHashMap<ResponseRequest> responseRequests = new IntHashMap<>();

   /**
    * Closed flag.
    */
   private final AtomicBoolean closed = new AtomicBoolean();


/*- Private final attributes (initialised by constructor) --------------------*/

   /**
    * The EPICS Channel-Access configuration variables to be used for this context.
    */
   private final ProtocolConfiguration protocolConfiguration;

   /**
    * Factory to be used for creating MonitorNotificationService instances.
    */
   private final MonitorNotificationServiceFactory monitorNotificationServiceFactory;

   /**
    * Repeater registration future.
    */
   private final ScheduledFuture<?> repeaterRegistrationFuture;

   /**
    * Leader/followers thread pool.
    */
   private final LeaderFollowersThreadPool leaderFollowersThreadPool;

   /**
    * Reactor.
    */
   private final Reactor reactor;

   /**
    * Channel search manager. Manages UDP search requests.
    */
   private final ChannelSearchManager channelSearchManager;

   /**
    * Cached hostname.
    */
   private final String hostName;

   /**
    * Cached username.
    */
   private final String userName;


/*- Private attributes --------------------------------------------------------*/

   /**
    * Last CID cache.
    */
   private int lastCID = 0;


   /**
    * Last IOID cache.
    */
   private int lastIOID = 0;


/*- Main ---------------------------------------------------------------------*/
/*- Constructor --------------------------------------------------------------*/

    /**
    * Creates an instance based on the supplied EPICS channel-access protocol
    * configuration.
    *
    * @param protocolConfiguration the configuration.
    * @throws NullPointerException if the properties argument was null.
    */
   public ContextImpl( ProtocolConfiguration protocolConfiguration )
   {
      Validate.notNull( protocolConfiguration, "null properties" );

      // Instantiate the protocol configuration object.
      this.protocolConfiguration = protocolConfiguration;

      // Capture the hostname and username.
      hostName = InetAddressUtil.getHostName();