Use Lambdas instead of Runnables, deprecate old inner classes
diff --git a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/auxiliary/lateral/socket/tcp/LateralTCPListener.java b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/auxiliary/lateral/socket/tcp/LateralTCPListener.java
index 3045253..8f457f8 100644
--- a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/auxiliary/lateral/socket/tcp/LateralTCPListener.java
+++ b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/auxiliary/lateral/socket/tcp/LateralTCPListener.java
@@ -19,7 +19,6 @@
* under the License.
*/
-import java.io.EOFException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
@@ -29,7 +28,6 @@
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
-import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Map;
import java.util.Set;
@@ -72,9 +70,6 @@
private static final ConcurrentHashMap<String, ILateralCacheListener<?, ?>> instances =
new ConcurrentHashMap<>();
- /** The socket listener */
- private ListenerThread receiver;
-
/** Configuration attributes */
private ITCPLateralCacheAttributes tcpLateralCacheAttributes;
@@ -122,8 +117,7 @@
newIns.init();
newIns.setCacheManager( cacheMgr );
- log.info( "Created new listener {0}",
- () -> ilca.getTcpListenerPort() );
+ log.info("Created new listener {0}", () -> ilca.getTcpListenerPort());
return newIns;
});
@@ -176,9 +170,7 @@
}
serverSocket.setSoTimeout( acceptTimeOut );
- receiver = new ListenerThread(serverSocket);
- receiver.setDaemon( true );
- receiver.start();
+ pooledExecutor.execute(() -> runListener(serverSocket));
}
catch ( final IOException ex )
{
@@ -444,7 +436,9 @@
/**
* Processes commands from the server socket. There should be one listener for each configured
* TCP lateral.
+ * @deprecated No longer used
*/
+ @Deprecated
public class ListenerThread
extends Thread
{
@@ -462,60 +456,62 @@
}
/** Main processing method for the ListenerThread object */
- @SuppressWarnings("synthetic-access")
@Override
public void run()
{
- try (ServerSocket ssck = serverSocket)
+ runListener(serverSocket);
+ }
+ }
+
+ /**
+ * Processes commands from the server socket. There should be one listener for each configured
+ * TCP lateral.
+ */
+ private void runListener(final ServerSocket serverSocket)
+ {
+ try
+ {
+ while ( true )
{
- ConnectionHandler handler;
+ log.debug( "Waiting for clients to connect " );
- outer: while ( true )
+ // Check to see if we've been asked to exit, and exit
+ if (terminated.get())
{
- log.debug( "Waiting for clients to connect " );
+ log.debug("Thread terminated, exiting gracefully");
+ break;
+ }
- Socket socket = null;
- inner: while (true)
+ try
+ {
+ final Socket socket = serverSocket.accept();
+
+ if (socket != null)
{
- // Check to see if we've been asked to exit, and exit
- if (terminated.get())
- {
- log.debug("Thread terminated, exiting gracefully");
- break outer;
- }
-
- try
- {
- socket = ssck.accept();
- break inner;
- }
- catch (final SocketTimeoutException e)
- {
- // No problem! We loop back up!
- continue inner;
- }
+ log.debug("Connected to client at {0}", () -> socket.getInetAddress());
}
- if ( socket != null && log.isDebugEnabled() )
- {
- final InetAddress inetAddress = socket.getInetAddress();
- log.debug( "Connected to client at {0}", inetAddress );
- }
-
- handler = new ConnectionHandler( socket );
- pooledExecutor.execute( handler );
+ pooledExecutor.execute(() -> handleConnection(socket));
+ }
+ catch (final SocketTimeoutException e)
+ {
+ // No problem! We loop back up!
}
}
- catch ( final IOException e )
- {
- log.error( "Exception caught in TCP listener", e );
- }
+
+ serverSocket.close();
+ }
+ catch ( final IOException e )
+ {
+ log.error( "Exception caught in TCP listener", e );
}
}
/**
* A Separate thread that runs when a command comes into the LateralTCPReceiver.
+ * @deprecated No longer used
*/
+ @Deprecated
public class ConnectionHandler
implements Runnable
{
@@ -535,125 +531,127 @@
* Main processing method for the LateralTCPReceiverConnection object
*/
@Override
- @SuppressWarnings({"unchecked", // Need to cast from Object
- "synthetic-access" })
public void run()
{
- try (ObjectInputStream ois =
- new ObjectInputStreamClassLoaderAware( socket.getInputStream(), null ))
+ handleConnection(socket);
+ }
+ }
+
+ /**
+ * A Separate thread that runs when a command comes into the LateralTCPReceiver.
+ */
+ private void handleConnection(final Socket socket)
+ {
+ try (ObjectInputStream ois =
+ new ObjectInputStreamClassLoaderAware( socket.getInputStream(), null ))
+ {
+ while ( true )
{
- while ( true )
+ @SuppressWarnings("unchecked") // Need to cast from Object
+ final LateralElementDescriptor<K, V> led =
+ (LateralElementDescriptor<K, V>) ois.readObject();
+
+ if ( led == null )
{
- final LateralElementDescriptor<K, V> led =
- (LateralElementDescriptor<K, V>) ois.readObject();
+ log.debug( "LateralElementDescriptor is null" );
+ continue;
+ }
+ if ( led.requesterId == getListenerId() )
+ {
+ log.debug( "from self" );
+ }
+ else
+ {
+ log.debug( "receiving LateralElementDescriptor from another led = {0}",
+ led );
- if ( led == null )
- {
- log.debug( "LateralElementDescriptor is null" );
- continue;
- }
- if ( led.requesterId == getListenerId() )
- {
- log.debug( "from self" );
- }
- else
- {
- log.debug( "receiving LateralElementDescriptor from another led = {0}",
- led );
-
- handle( led );
- }
+ handleElement(led, socket);
}
}
- catch ( final EOFException e )
- {
- log.info( "Caught EOFException, closing connection.", e );
- }
- catch ( final SocketException e )
- {
- log.info( "Caught SocketException, closing connection.", e );
- }
- catch ( final Exception e )
- {
- log.error( "Unexpected exception.", e );
- }
}
-
- /**
- * This calls the appropriate method, based on the command sent in the Lateral element
- * descriptor.
- * <p>
- * @param led
- * @throws IOException
- */
- @SuppressWarnings("synthetic-access")
- private void handle( final LateralElementDescriptor<K, V> led )
- throws IOException
+ catch (final IOException e)
{
- final String cacheName = led.ce.getCacheName();
- final K key = led.ce.getKey();
- Serializable obj = null;
+ log.info("Caught {0}, closing connection.", e.getClass().getSimpleName(), e);
+ }
+ catch (final ClassNotFoundException e)
+ {
+ log.error( "Deserialization failed reading from socket", e );
+ }
+ }
- switch (led.command)
- {
- case UPDATE:
- handlePut( led.ce );
- break;
+ /**
+ * This calls the appropriate method, based on the command sent in the Lateral element
+ * descriptor.
+ * <p>
+ * @param led the lateral element
+ * @param socket the socket
+ * @throws IOException
+ */
+ private void handleElement(final LateralElementDescriptor<K, V> led, Socket socket) throws IOException
+ {
+ final String cacheName = led.ce.getCacheName();
+ final K key = led.ce.getKey();
+ Serializable obj = null;
- case REMOVE:
- // if a hashcode was given and filtering is on
- // check to see if they are the same
- // if so, then don't remove, otherwise issue a remove
- if ( led.valHashCode != -1 )
+ switch (led.command)
+ {
+ case UPDATE:
+ handlePut( led.ce );
+ break;
+
+ case REMOVE:
+ // if a hashcode was given and filtering is on
+ // check to see if they are the same
+ // if so, then don't remove, otherwise issue a remove
+ if ( led.valHashCode != -1 )
+ {
+ if ( getTcpLateralCacheAttributes().isFilterRemoveByHashCode() )
{
- if ( getTcpLateralCacheAttributes().isFilterRemoveByHashCode() )
+ final ICacheElement<K, V> test = getCache( cacheName ).localGet( key );
+ if ( test != null )
{
- final ICacheElement<K, V> test = getCache( cacheName ).localGet( key );
- if ( test != null )
+ if ( test.getVal().hashCode() == led.valHashCode )
{
- if ( test.getVal().hashCode() == led.valHashCode )
- {
- log.debug( "Filtering detected identical hashCode [{0}], "
- + "not issuing a remove for led {1}",
- led.valHashCode, led );
- return;
- }
- else
- {
- log.debug( "Different hashcodes, in cache [{0}] sent [{1}]",
- test.getVal().hashCode(), led.valHashCode );
- }
+ log.debug( "Filtering detected identical hashCode [{0}], "
+ + "not issuing a remove for led {1}",
+ led.valHashCode, led );
+ return;
+ }
+ else
+ {
+ log.debug( "Different hashcodes, in cache [{0}] sent [{1}]",
+ test.getVal().hashCode(), led.valHashCode );
}
}
}
- handleRemove( cacheName, key );
- break;
+ }
+ handleRemove( cacheName, key );
+ break;
- case REMOVEALL:
- handleRemoveAll( cacheName );
- break;
+ case REMOVEALL:
+ handleRemoveAll( cacheName );
+ break;
- case GET:
- obj = handleGet( cacheName, key );
- break;
+ case GET:
+ obj = handleGet( cacheName, key );
+ break;
- case GET_MATCHING:
- obj = (Serializable) handleGetMatching( cacheName, (String) key );
- break;
+ case GET_MATCHING:
+ obj = (Serializable) handleGetMatching( cacheName, (String) key );
+ break;
- case GET_KEYSET:
- obj = (Serializable) handleGetKeySet(cacheName);
- break;
+ case GET_KEYSET:
+ obj = (Serializable) handleGetKeySet(cacheName);
+ break;
- default: break;
- }
+ default: break;
+ }
- if (obj != null)
- {
- final ObjectOutputStream oos = new ObjectOutputStream( socket.getOutputStream() );
- oos.writeObject( obj );
- oos.flush();
- }
+ if (obj != null)
+ {
+ final ObjectOutputStream oos = new ObjectOutputStream( socket.getOutputStream() );
+ oos.writeObject( obj );
+ oos.flush();
}
}
@@ -666,8 +664,7 @@
if ( shutdown.compareAndSet(false, true) )
{
log.info( "Shutting down TCP Lateral receiver." );
-
- receiver.interrupt();
+ pooledExecutor.shutdownNow();
}
else
{
diff --git a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/AbstractCacheEventQueue.java b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/AbstractCacheEventQueue.java
index 6055a82..f131313 100644
--- a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/AbstractCacheEventQueue.java
+++ b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/AbstractCacheEventQueue.java
@@ -205,45 +205,40 @@
*/
protected abstract class AbstractCacheEvent implements Runnable
{
- /** Number of failures encountered processing this event. */
- int failures;
-
/**
* Main processing method for the AbstractCacheEvent object
*/
@Override
- @SuppressWarnings("synthetic-access")
public void run()
{
- try
+ for (int failures = 0; failures < maxFailure; failures++)
{
- doRun();
- }
- catch ( final IOException e )
- {
- log.warn( e );
- if ( ++failures >= maxFailure )
+ try
{
- log.warn( "Error while running event from Queue: {0}. "
- + "Dropping Event and marking Event Queue as "
- + "non-functional.", this );
- destroy();
+ doRun();
return;
}
- log.info( "Error while running event from Queue: {0}. "
- + "Retrying...", this );
+ catch (final IOException e)
+ {
+ log.warn("Error while running event from Queue: {0}. "
+ + "Retrying...", this, e);
+ }
+
try
{
Thread.sleep( waitBeforeRetry );
- run();
}
catch ( final InterruptedException ie )
{
- log.warn( "Interrupted while sleeping for retry on event "
- + "{0}.", this );
- destroy();
+ log.warn("Interrupted while sleeping for retry on event "
+ + "{0}.", this, ie);
+ break;
}
}
+
+ log.warn( "Dropping Event and marking Event Queue {0} as "
+ + "non-functional.", this );
+ destroy();
}
/**
diff --git a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/control/event/ElementEventQueue.java b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/control/event/ElementEventQueue.java
index e59c611..fd85be2 100644
--- a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/control/event/ElementEventQueue.java
+++ b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/engine/control/event/ElementEventQueue.java
@@ -21,6 +21,7 @@
import java.io.IOException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.jcs3.engine.control.event.behavior.IElementEvent;
import org.apache.commons.jcs3.engine.control.event.behavior.IElementEventHandler;
@@ -28,8 +29,8 @@
import org.apache.commons.jcs3.log.Log;
import org.apache.commons.jcs3.log.LogManager;
import org.apache.commons.jcs3.utils.threadpool.PoolConfiguration;
-import org.apache.commons.jcs3.utils.threadpool.ThreadPoolManager;
import org.apache.commons.jcs3.utils.threadpool.PoolConfiguration.WhenBlockedPolicy;
+import org.apache.commons.jcs3.utils.threadpool.ThreadPoolManager;
/**
* An event queue is used to propagate ordered cache events to one and only one target listener.
@@ -43,7 +44,7 @@
private static final Log log = LogManager.getLog( ElementEventQueue.class );
/** shutdown or not */
- private boolean destroyed;
+ private AtomicBoolean destroyed = new AtomicBoolean(false);
/** The worker thread pool. */
private ExecutorService queueProcessor;
@@ -65,15 +66,10 @@
@Override
public void dispose()
{
- if ( !destroyed )
+ if (destroyed.compareAndSet(false, true))
{
- destroyed = true;
-
- // synchronize on queue so the thread will not wait forever,
- // and then interrupt the QueueProcessor
+ // shut down the QueueProcessor
queueProcessor.shutdownNow();
- queueProcessor = null;
-
log.info( "Element event queue destroyed: {0}", this );
}
}
@@ -89,19 +85,15 @@
throws IOException
{
- log.debug( "Adding Event Handler to QUEUE, !destroyed = {0}", !destroyed );
+ log.debug("Adding Event Handler to QUEUE, !destroyed = {0}", !destroyed.get());
- if (destroyed)
+ if (destroyed.get())
{
log.warn("Event submitted to disposed element event queue {0}", event);
}
else
{
- final ElementEventRunner runner = new ElementEventRunner( hand, event );
-
- log.debug( "runner = {0}", runner );
-
- queueProcessor.execute(runner);
+ queueProcessor.execute(() -> hand.handleElementEvent(event));
}
}
@@ -109,14 +101,15 @@
/**
* Retries before declaring failure.
+ * @deprecated No longer used
*/
+ @Deprecated
protected abstract class AbstractElementEventRunner
implements Runnable
{
/**
* Main processing method for the AbstractElementEvent object
*/
- @SuppressWarnings("synthetic-access")
@Override
public void run()
{
@@ -140,45 +133,4 @@
protected abstract void doRun()
throws IOException;
}
-
- /**
- * ElementEventRunner.
- */
- private class ElementEventRunner
- extends AbstractElementEventRunner
- {
- /** the handler */
- private final IElementEventHandler hand;
-
- /** event */
- private final IElementEvent<?> event;
-
- /**
- * Constructor for the PutEvent object.
- * <p>
- * @param hand
- * @param event
- * @throws IOException
- */
- @SuppressWarnings("synthetic-access")
- ElementEventRunner( final IElementEventHandler hand, final IElementEvent<?> event )
- throws IOException
- {
- log.debug( "Constructing {0}", this );
- this.hand = hand;
- this.event = event;
- }
-
- /**
- * Tells the handler to handle the event.
- * <p>
- * @throws IOException
- */
- @Override
- protected void doRun()
- throws IOException
- {
- hand.handleElementEvent( event );
- }
- }
}
diff --git a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPCleanupRunner.java b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPCleanupRunner.java
index e94f5ac..647dc2d 100644
--- a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPCleanupRunner.java
+++ b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPCleanupRunner.java
@@ -19,9 +19,6 @@
* under the License.
*/
-import java.util.HashSet;
-import java.util.Set;
-
import org.apache.commons.jcs3.log.Log;
import org.apache.commons.jcs3.log.LogManager;
@@ -66,27 +63,20 @@
{
final long now = System.currentTimeMillis();
- // iterate through the set
- // it is thread safe
- // TODO this should get a copy. you can't simply remove from this.
// the listeners need to be notified.
- final Set<DiscoveredService> toRemove = new HashSet<>();
- // can't remove via the iterator. must remove directly
- for (final DiscoveredService service : discoveryService.getDiscoveredServices())
- {
- if ( ( now - service.getLastHearFromTime() ) > ( maxIdleTimeSeconds * 1000 ) )
- {
- log.info( "Removing service, since we haven't heard from it in "
- + "{0} seconds. service = {1}", maxIdleTimeSeconds, service );
- toRemove.add( service );
- }
- }
+ discoveryService.getDiscoveredServices().stream()
+ .filter(service -> {
+ if (now - service.getLastHearFromTime() > maxIdleTimeSeconds * 1000)
+ {
+ log.info( "Removing service, since we haven't heard from it in "
+ + "{0} seconds. service = {1}", maxIdleTimeSeconds, service );
+ return true;
+ }
- // remove the bad ones
- for (final DiscoveredService service : toRemove)
- {
+ return false;
+ })
+ // remove the bad ones
// call this so the listeners get notified
- discoveryService.removeDiscoveredService( service );
- }
+ .forEach(service -> discoveryService.removeDiscoveredService(service));
}
}
diff --git a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPDiscoveryReceiver.java b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPDiscoveryReceiver.java
index 3f03ca5..5e71d63 100644
--- a/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPDiscoveryReceiver.java
+++ b/commons-jcs-core/src/main/java/org/apache/commons/jcs3/utils/discovery/UDPDiscoveryReceiver.java
@@ -37,8 +37,8 @@
import org.apache.commons.jcs3.utils.discovery.UDPDiscoveryMessage.BroadcastType;
import org.apache.commons.jcs3.utils.net.HostNameUtil;
import org.apache.commons.jcs3.utils.threadpool.PoolConfiguration;
-import org.apache.commons.jcs3.utils.threadpool.ThreadPoolManager;
import org.apache.commons.jcs3.utils.threadpool.PoolConfiguration.WhenBlockedPolicy;
+import org.apache.commons.jcs3.utils.threadpool.ThreadPoolManager;
/** Receives UDP Discovery messages. */
public class UDPDiscoveryReceiver
@@ -210,18 +210,13 @@
log.debug( "{0} messages received.", this::getCnt );
- UDPDiscoveryMessage message = null;
-
try
{
- message = (UDPDiscoveryMessage) obj;
+ UDPDiscoveryMessage message = (UDPDiscoveryMessage) obj;
// check for null
if ( message != null )
{
- final MessageHandler handler = new MessageHandler( message );
-
- pooledExecutor.execute( handler );
-
+ pooledExecutor.execute(() -> handleMessage(message));
log.debug( "Passed handler to executor." );
}
else
@@ -269,7 +264,9 @@
/**
* Separate thread run when a command comes into the UDPDiscoveryReceiver.
+ * @deprectaed No longer used
*/
+ @Deprecated
public class MessageHandler
implements Runnable
{
@@ -287,62 +284,69 @@
/**
* Process the message.
*/
- @SuppressWarnings("synthetic-access")
@Override
public void run()
{
- // consider comparing ports here instead.
- if ( message.getRequesterId() == CacheInfo.listenerId )
- {
- log.debug( "Ignoring message sent from self" );
- }
- else
- {
- log.debug( "Process message sent from another" );
- log.debug( "Message = {0}", message );
-
- if ( message.getHost() == null || message.getCacheNames() == null || message.getCacheNames().isEmpty() )
- {
- log.debug( "Ignoring invalid message: {0}", message );
- }
- else
- {
- processMessage();
- }
- }
+ handleMessage(message);
}
- /**
- * Process the incoming message.
- */
- @SuppressWarnings("synthetic-access")
- private void processMessage()
- {
- final DiscoveredService discoveredService = new DiscoveredService();
- discoveredService.setServiceAddress( message.getHost() );
- discoveredService.setCacheNames( message.getCacheNames() );
- discoveredService.setServicePort( message.getPort() );
- discoveredService.setLastHearFromTime( System.currentTimeMillis() );
+ }
- // if this is a request message, have the service handle it and
- // return
- if ( message.getMessageType() == BroadcastType.REQUEST )
+ /**
+ * Separate thread run when a command comes into the UDPDiscoveryReceiver.
+ */
+ private void handleMessage(UDPDiscoveryMessage message)
+ {
+ // consider comparing ports here instead.
+ if ( message.getRequesterId() == CacheInfo.listenerId )
+ {
+ log.debug( "Ignoring message sent from self" );
+ }
+ else
+ {
+ log.debug( "Process message sent from another" );
+ log.debug( "Message = {0}", message );
+
+ if ( message.getHost() == null || message.getCacheNames() == null || message.getCacheNames().isEmpty() )
{
- log.debug( "Message is a Request Broadcast, will have the service handle it." );
- service.serviceRequestBroadcast();
- }
- else if ( message.getMessageType() == BroadcastType.REMOVE )
- {
- log.debug( "Removing service from set {0}", discoveredService );
- service.removeDiscoveredService( discoveredService );
+ log.debug( "Ignoring invalid message: {0}", message );
}
else
{
- service.addOrUpdateService( discoveredService );
+ processMessage(message);
}
}
}
+ /**
+ * Process the incoming message.
+ */
+ private void processMessage(UDPDiscoveryMessage message)
+ {
+ final DiscoveredService discoveredService = new DiscoveredService();
+ discoveredService.setServiceAddress( message.getHost() );
+ discoveredService.setCacheNames( message.getCacheNames() );
+ discoveredService.setServicePort( message.getPort() );
+ discoveredService.setLastHearFromTime( System.currentTimeMillis() );
+
+ // if this is a request message, have the service handle it and
+ // return
+ if ( message.getMessageType() == BroadcastType.REQUEST )
+ {
+ log.debug( "Message is a Request Broadcast, will have the service handle it." );
+ service.serviceRequestBroadcast();
+ }
+ else if ( message.getMessageType() == BroadcastType.REMOVE )
+ {
+ log.debug( "Removing service from set {0}", discoveredService );
+ service.removeDiscoveredService( discoveredService );
+ }
+ else
+ {
+ service.addOrUpdateService( discoveredService );
+ }
+ }
+
/** Shuts down the socket. */
@Override
public void shutdown()
