Post migration to JDK 8: Functions, lambdas and use of computeIfAbsent to behave like a true cache

Author: Guido Medina

quickfixj-codegenerator/src/main/java/org/quickfixj/codegenerator/MessageCodeGenerator.java

@@ -241,7 +241,7 @@ private void writePackageDocumentation(String outputDirectory, String descriptio
     }
 
     private List<String> getNames(Element element, String path) {
-        return getNames(element, path, new ArrayList<String>());
+        return getNames(element, path, new ArrayList<>());
     }
 
     private List<String> getNames(Element element, String path, List<String> names) {

quickfixj-core/src/main/java/org/quickfixj/SimpleCache.java

@@ -0,0 +1,42 @@
+/*******************************************************************************
+ * Copyright (c) quickfixengine.org  All rights reserved.
+ *
+ * This file is part of the QuickFIX FIX Engine
+ *
+ * This file may be distributed under the terms of the quickfixengine.org
+ * license as defined by quickfixengine.org and appearing in the file
+ * LICENSE included in the packaging of this file.
+ *
+ * This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING
+ * THE WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A
+ * PARTICULAR PURPOSE.
+ *
+ * See http://www.quickfixengine.org/LICENSE for licensing information.
+ *
+ * Contact ask@quickfixengine.org if any conditions of this licensing
+ * are not clear to you.
+ ******************************************************************************/
+
+package org.quickfixj;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+public class SimpleCache<K, V> extends ConcurrentHashMap<K, V> {
+    static final int CPUs = Runtime.getRuntime().availableProcessors();
+
+    final Function<K, V> loadingFunction;
+
+    public SimpleCache(Function<K, V> loadingFunction) {
+        super(CPUs, 0.7f, CPUs);
+        this.loadingFunction = loadingFunction;
+    }
+
+    public V computeIfAbsent(K key) {
+        /*
+         * We could computeIfAbsent directly but for CPUs < 32 pre-scanning is faster.
+         */
+        final V value = get(key);
+        return value != null ? value : computeIfAbsent(key, loadingFunction);
+    }
+}

quickfixj-core/src/main/java/org/quickfixj/jmx/mbean/connector/ConnectorAdmin.java

@@ -37,8 +37,6 @@
 import javax.management.ObjectName;
 import javax.management.openmbean.OpenDataException;
 import javax.management.openmbean.TabularData;
-import java.beans.PropertyChangeEvent;
-import java.beans.PropertyChangeListener;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
@@ -167,11 +165,9 @@ public ObjectName preRegister(MBeanServer server, ObjectName name) throws Except
 
     public void postRegister(Boolean registrationDone) {
         if (connector instanceof SessionConnector) {
-            ((SessionConnector) connector).addPropertyChangeListener(new PropertyChangeListener() {
-                public void propertyChange(PropertyChangeEvent evt) {
-                    if (SessionConnector.SESSIONS_PROPERTY.equals(evt.getPropertyName())) {
-                        registerSessions();
-                    }
+            ((SessionConnector) connector).addPropertyChangeListener(evt -> {
+                if (SessionConnector.SESSIONS_PROPERTY.equals(evt.getPropertyName())) {
+                    registerSessions();
                 }
             });
         }

quickfixj-core/src/main/java/quickfix/DefaultDataDictionaryProvider.java

@@ -20,54 +20,51 @@
 package quickfix;
 
 import org.quickfixj.QFJException;
+import org.quickfixj.SimpleCache;
 import quickfix.field.ApplVerID;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
 import static quickfix.MessageUtils.toBeginString;
 
 public class DefaultDataDictionaryProvider implements DataDictionaryProvider {
-    private final Map<String, DataDictionary> transportDictionaries = new ConcurrentHashMap<>();
-    private final Map<AppVersionKey, DataDictionary> applicationDictionaries = new ConcurrentHashMap<>();
-    private final boolean findDataDictionaries;
+    private final SimpleCache<String, DataDictionary> transportDictionaries;
+    private final SimpleCache<AppVersionKey, DataDictionary> applicationDictionaries;
 
     public DefaultDataDictionaryProvider() {
-        findDataDictionaries = true;
+        this(true);
     }
 
     public DefaultDataDictionaryProvider(boolean findDataDictionaries) {
-        this.findDataDictionaries = findDataDictionaries;
+        transportDictionaries = new SimpleCache<>(beginString -> {
+            if (findDataDictionaries) {
+                final String path = beginString.replace(".", "") + ".xml";
+                try {
+                    return new DataDictionary(path);
+                } catch (ConfigError e) {
+                    throw new QFJException(e);
+                }
+            }
+            return null;
+        });
+        applicationDictionaries = new SimpleCache<>(appVersionKey -> {
+            if (findDataDictionaries) {
+                final String beginString = toBeginString(appVersionKey.applVerID);
+                final String path = beginString.replace(".", "") + ".xml";
+                try {
+                    return new DataDictionary(path);
+                } catch (ConfigError e) {
+                    throw new QFJException(e);
+                }
+            }
+            return null;
+        });
     }
 
-    public synchronized DataDictionary getSessionDataDictionary(String beginString) {
-        DataDictionary dd = transportDictionaries.get(beginString);
-        if (dd == null && findDataDictionaries) {
-            String path = beginString.replace(".", "") + ".xml";
-            try {
-                dd = new DataDictionary(path);
-                transportDictionaries.put(beginString, dd);
-            } catch (ConfigError e) {
-                throw new QFJException(e);
-            }
-        }
-        return dd;
+    public DataDictionary getSessionDataDictionary(String beginString) {
+        return transportDictionaries.computeIfAbsent(beginString);
     }
 
     public DataDictionary getApplicationDataDictionary(ApplVerID applVerID) {
-        AppVersionKey appVersionKey = new AppVersionKey(applVerID);
-        DataDictionary dd = applicationDictionaries.get(appVersionKey);
-        if (dd == null && findDataDictionaries) {
-            String beginString = toBeginString(applVerID);
-            String path = beginString.replace(".", "") + ".xml";
-            try {
-                dd = new DataDictionary(path);
-                applicationDictionaries.put(appVersionKey, dd);
-            } catch (ConfigError e) {
-                throw new QFJException(e);
-            }
-        }
-        return dd;
+        return applicationDictionaries.computeIfAbsent(new AppVersionKey(applVerID));
     }
 
     public void addTransportDictionary(String beginString, DataDictionary dd) {

quickfixj-core/src/main/java/quickfix/DefaultSessionFactory.java

@@ -19,22 +19,29 @@
 
 package quickfix;
 
+import org.quickfixj.QFJException;
+import org.quickfixj.SimpleCache;
 import quickfix.field.ApplVerID;
 import quickfix.field.DefaultApplVerID;
 
 import java.net.InetAddress;
 import java.util.Enumeration;
-import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Factory for creating sessions. Used by the communications code (acceptors,
  * initiators) for creating sessions.
  */
 public class DefaultSessionFactory implements SessionFactory {
-    private static final Map<String, DataDictionary> dictionaryCache = new ConcurrentHashMap<>();
+    private static final SimpleCache<String, DataDictionary> dictionaryCache = new SimpleCache<>(path -> {
+        try {
+            return new DataDictionary(path);
+        } catch (ConfigError e) {
+            throw new QFJException(e.getMessage(), e);
+        }
+    });
+
     private final Application application;
     private final MessageStoreFactory messageStoreFactory;
     private final LogFactory logFactory;
@@ -320,14 +327,7 @@ private String toDictionaryPath(String beginString) {
     }
 
     private DataDictionary getDataDictionary(String path) throws ConfigError {
-        synchronized (dictionaryCache) {
-            DataDictionary dataDictionary = dictionaryCache.get(path);
-            if (dataDictionary == null) {
-                dataDictionary = new DataDictionary(path);
-                dictionaryCache.put(path, dataDictionary);
-            }
-            return dataDictionary;
-        }
+        return dictionaryCache.computeIfAbsent(path);
     }
 
     private int[] getLogonIntervalsInSeconds(SessionSettings settings, SessionID sessionID) throws ConfigError {

quickfixj-core/src/main/java/quickfix/FileStore.java

@@ -76,7 +76,7 @@ public class FileStore implements MessageStore, Closeable {
         this.syncWrites = syncWrites;
         this.maxCachedMsgs = maxCachedMsgs;
 
-        messageIndex = maxCachedMsgs > 0 ? new TreeMap<Long, long[]>() : null;
+        messageIndex = maxCachedMsgs > 0 ? new TreeMap<>() : null;
 
         final String fullPath = new File(path == null ? "." : path).getAbsolutePath();
         final String sessionName = FileUtil.sessionIdFileName(sessionID);

quickfixj-core/src/main/java/quickfix/JdbcUtil.java

@@ -31,13 +31,15 @@
 import java.sql.SQLException;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
 
 class JdbcUtil {
 
     static final String CONNECTION_POOL_ALIAS = "quickfixj";
 
     private static final Map<String, ProxoolDataSource> dataSources = new ConcurrentHashMap<>();
-    private static int dataSourceCounter = 1;
+    private static AtomicInteger dataSourceCounter = new AtomicInteger();
 
     static DataSource getDataSource(SessionSettings settings, SessionID sessionID)
             throws ConfigError, FieldConvertError {
@@ -81,35 +83,35 @@ static DataSource getDataSource(String jdbcDriver, String connectionURL, String
     }
 
     /**
-     * This is typically called from a single thread, but just in case we are synchronizing modification
-     * of the cache. The cache itself is thread safe.
+     * This is typically called from a single thread, but just in case we are using an atomic loading function
+     * to avoid the creation of two data sources simultaneously. The cache itself is thread safe.
      */
-    static synchronized DataSource getDataSource(String jdbcDriver, String connectionURL, String user, String password,
+    static DataSource getDataSource(String jdbcDriver, String connectionURL, String user, String password,
                                                  boolean cache, int maxConnCount, int simultaneousBuildThrottle,
                                                  long maxActiveTime, int maxConnLifetime) {
         String key = jdbcDriver + "#" + connectionURL + "#" + user + "#" + password;
         ProxoolDataSource ds = cache ? dataSources.get(key) : null;
 
         if (ds == null) {
-            ds = new ProxoolDataSource(JdbcUtil.CONNECTION_POOL_ALIAS + "-" + dataSourceCounter++);
-
-            ds.setDriver(jdbcDriver);
-            ds.setDriverUrl(connectionURL);
-
-            // Bug in Proxool 0.9RC2. Must set both delegate properties and individual setters. :-(
-            ds.setDelegateProperties("user=" + user + ","
-                    + (password != null && !"".equals(password) ? "password=" + password : ""));
-            ds.setUser(user);
-            ds.setPassword(password);
-
-            ds.setMaximumActiveTime(maxActiveTime);
-            ds.setMaximumConnectionLifetime(maxConnLifetime);
-            ds.setMaximumConnectionCount(maxConnCount);
-            ds.setSimultaneousBuildThrottle(simultaneousBuildThrottle);
-
-            if (cache) {
-                dataSources.put(key, ds);
-            }
+            final Function<String, ProxoolDataSource> loadingFunction = dataSourceKey -> {
+                final ProxoolDataSource dataSource = new ProxoolDataSource(CONNECTION_POOL_ALIAS + "-" + dataSourceCounter.incrementAndGet());
+
+                dataSource.setDriver(jdbcDriver);
+                dataSource.setDriverUrl(connectionURL);
+
+                // Bug in Proxool 0.9RC2. Must set both delegate properties and individual setters. :-(
+                dataSource.setDelegateProperties("user=" + user + ","
+                        + (password != null && !"".equals(password) ? "password=" + password : ""));
+                dataSource.setUser(user);
+                dataSource.setPassword(password);
+
+                dataSource.setMaximumActiveTime(maxActiveTime);
+                dataSource.setMaximumConnectionLifetime(maxConnLifetime);
+                dataSource.setMaximumConnectionCount(maxConnCount);
+                dataSource.setSimultaneousBuildThrottle(simultaneousBuildThrottle);
+                return dataSource;
+            };
+            ds = cache ? dataSources.computeIfAbsent(key, loadingFunction) : loadingFunction.apply(key);
         }
         return ds;
     }

quickfixj-core/src/main/java/quickfix/SystemTime.java

@@ -29,11 +29,7 @@
 public class SystemTime {
     public static final TimeZone UTC_TIMEZONE = TimeZone.getTimeZone("UTC");
 
-    private static final SystemTimeSource DEFAULT_TIME_SOURCE = new SystemTimeSource() {
-        public long getTime() {
-            return System.currentTimeMillis();
-        }
-    };
+    private static final SystemTimeSource DEFAULT_TIME_SOURCE = System::currentTimeMillis;
 
     private static volatile SystemTimeSource systemTimeSource = DEFAULT_TIME_SOURCE;
 

quickfixj-core/src/main/java/quickfix/field/converter/UtcTimestampConverter.java

@@ -19,24 +19,33 @@
 
 package quickfix.field.converter;
 
+import org.quickfixj.SimpleCache;
 import quickfix.FieldConvertError;
 import quickfix.SystemTime;
 
 import java.text.DateFormat;
 import java.util.Calendar;
 import java.util.Date;
 import java.util.GregorianCalendar;
-import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Convert between a timestamp and a String. A timestamp includes both a date
  * and a time.
  */
 public class UtcTimestampConverter extends AbstractDateTimeConverter {
     private static final ThreadLocal<UtcTimestampConverter> utcTimestampConverter = new ThreadLocal<>();
+    private final static SimpleCache<String, Long> dateCache = new SimpleCache<>(dateString -> {
+        final Calendar c = new GregorianCalendar(1970, 0, 1, 0, 0, 0);
+        c.setTimeZone(SystemTime.UTC_TIMEZONE);
+        final int year = Integer.parseInt(dateString.substring(0, 4));
+        final int month = Integer.parseInt(dateString.substring(4, 6));
+        final int day = Integer.parseInt(dateString.substring(6, 8));
+        c.set(year, month - 1, day);
+        return c.getTimeInMillis();
+    });
+
     private final DateFormat utcTimestampFormat = createDateFormat("yyyyMMdd-HH:mm:ss");
     private final DateFormat utcTimestampFormatMillis = createDateFormat("yyyyMMdd-HH:mm:ss.SSS");
-    private final static ConcurrentHashMap<String, Long> dateCache = new ConcurrentHashMap<>();
 
     /**
      * Convert a timestamp (represented as a Date) to a String.
@@ -83,19 +92,7 @@ public static Date convert(String value) throws FieldConvertError {
     }
 
     private static Long getMillisForDay(String value) {
-        String dateString = value.substring(0, 8);
-        Long millis = dateCache.get(dateString);
-        if (millis == null) {
-            Calendar c = new GregorianCalendar(1970, 0, 1, 0, 0, 0);
-            c.setTimeZone(SystemTime.UTC_TIMEZONE);
-            int year = Integer.parseInt(value.substring(0, 4));
-            int month = Integer.parseInt(value.substring(4, 6));
-            int day = Integer.parseInt(value.substring(6, 8));
-            c.set(year, month - 1, day);
-            millis = c.getTimeInMillis();
-            dateCache.put(dateString, c.getTimeInMillis());
-        }
-        return millis;
+        return dateCache.computeIfAbsent(value.substring(0, 8));
     }
 
     private static void verifyFormat(String value) throws FieldConvertError {

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

@@ -121,13 +121,10 @@ public void blockInThread() {
         }
 
         startHandlingMessages();
-        messageProcessingThread = new Thread(new Runnable() {
-            @Override
-            public void run() {
-                sessionConnector.log.info("Started " + MESSAGE_PROCESSOR_THREAD_NAME);
-                block();
-                sessionConnector.log.info("Stopped " + MESSAGE_PROCESSOR_THREAD_NAME);
-            }
+        messageProcessingThread = new Thread(() -> {
+            sessionConnector.log.info("Started " + MESSAGE_PROCESSOR_THREAD_NAME);
+            block();
+            sessionConnector.log.info("Stopped " + MESSAGE_PROCESSOR_THREAD_NAME);
         }, MESSAGE_PROCESSOR_THREAD_NAME);
         messageProcessingThread.setDaemon(true);
         messageProcessingThread.start();