Keep unsent messages when the session is reset by its schedule

quickfixj-core/src/main/doc/usermanual/usage/configuration.md

@@ -217,6 +217,7 @@ Acceptor and Initiator socket options can be set in either defaults or per-sessi
 | `JdbcPassword` | JDBC password. Also used for JDBC log. | | |
 | `JdbcStoreMessagesTableName` | Table name for messages table. | A valid SQL table name. | `messages` |
 | `JdbcStoreSessionsTableName` | Table name for sessions table. | A valid SQL table name. | `sessions` |
+| `JdbcStoreUnsentSeqNumsTableName` | Table name for the table that keeps the first unsent sequence number. This table is optional. It is only needed for `KeepUnsentMessagesOnReset`. | A valid SQL table name. | `unsent_seqnums` |
 | `JdbcLogHeartBeats` | Controls filtering of heartbeats for message logging (both in and out). | `Y`<br/>`N` | `N` |
 | `JdbcLogIncomingTable` | The name of the JDBC log incoming table. | valid table name | `messages_log` |
 | `JdbcLogOutgoingTable` | The name of the JDBC log outgoing table. | valid table name | `messages_log` |
@@ -277,6 +278,7 @@ Acceptor and Initiator socket options can be set in either defaults or per-sessi
 | `HeartBeatTimeoutMultiplier` | Fraction of the heartbeat interval which defines the additional time to wait since the last message was received before disconnecting (final coefficient value is equal to `HeartBeatTimeoutMultiplier + 1.0`). | any non-negative value | `1.4` |
 | `DisableHeartBeatCheck` | Heartbeat detection is disabled. A disconnect due to a missing heartbeat will never occur. | `Y`<br/>`N` | `N` |
 | `ForceResendWhenCorruptedStore` | Fill in heartbeats on resend when reading from message store fails. | `Y`<br/>`N` | `N` |
+| `KeepUnsentMessagesOnReset` | Keep the application messages that were stored but not sent (for example sent outside of the session time) when the session is reset by its schedule. Without it, these messages are lost. The kept messages get new sequence numbers. They are sent again with `PossDupFlag=Y` when the counterparty asks for them after the next logon. Works only with `PersistMessages=Y` and with `MemoryStore`, `FileStore`, `CachedFileStore` or `JdbcStore`. `JdbcStore` needs the table set by `JdbcStoreUnsentSeqNumsTableName`. Messages are still lost with `ResetOnLogon`, `ResetOnLogout`, `ResetOnDisconnect`, a Logon with `ResetSeqNumFlag=Y` or a call to `Session.reset()`. | `Y`<br/>`N` | `N` |
 | `LogMessageWhenSessionNotFound` | Log the entire message when the corresponding session can not be found. Otherwise only the SessionID is logged. | `Y`<br/>`N` | `Y` |
 | `AllowPosDup` | Whether to allow `PossDupFlag` and `OrigSendingTime` when sending messages. This is useful on occasions, primarily when a QFJ application is acting as purely a pass-through/monitoring hop. | `Y`<br/>`N` | `N` |
 

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

@@ -73,6 +73,8 @@ public class CachedFileStore implements MessageStore {
 
     private final String sessionFileName;
 
+    private final String firstUnsentSeqNumFileName;
+
     private RandomAccessFile messageFileReader;
 
     private RandomAccessFile messageFileWriter;
@@ -81,6 +83,8 @@ public class CachedFileStore implements MessageStore {
 
     private RandomAccessFile sequenceNumberFile;
 
+    private RandomAccessFile firstUnsentSeqNumFile;
+
     private final boolean syncWrites;
 
     private final CachedHashMap messageIndex = new CachedHashMap(100);
@@ -98,6 +102,7 @@ public class CachedFileStore implements MessageStore {
         headerFileName = prefix + "header";
         seqNumFileName = prefix + "seqnums";
         sessionFileName = prefix + "session";
+        firstUnsentSeqNumFileName = prefix + "firstunsentseqnum";
 
         final File directory = new File(msgFileName).getParentFile();
         directory.mkdirs();
@@ -115,6 +120,7 @@ void initialize(boolean deleteFiles) throws IOException {
         messageFileWriter = new RandomAccessFile(msgFileName, getRandomAccessFileOptions());
         messageFileReader = new RandomAccessFile(msgFileName, READ_OPTION);
         sequenceNumberFile = new RandomAccessFile(seqNumFileName, getRandomAccessFileOptions());
+        firstUnsentSeqNumFile = new RandomAccessFile(firstUnsentSeqNumFileName, getRandomAccessFileOptions());
 
         initializeCache();
     }
@@ -180,6 +186,12 @@ private void initializeSequenceNumbers() throws IOException {
             cache.setNextSenderMsgSeqNum(Integer.parseInt(s.substring(0, offset)));
             cache.setNextTargetMsgSeqNum(Integer.parseInt(s.substring(offset + 1)));
         }
+
+        firstUnsentSeqNumFile.seek(0);
+        if (firstUnsentSeqNumFile.length() > 0) {
+            final String s = firstUnsentSeqNumFile.readUTF();
+            cache.setFirstUnsentSenderMsgSeqNum(Integer.parseInt(s));
+        }
     }
 
     private void initializeMessageIndex() throws IOException {
@@ -214,6 +226,7 @@ public void closeFiles() throws IOException {
         closeFile(messageFileWriter);
         closeFile(messageFileReader);
         closeFile(sequenceNumberFile);
+        closeFile(firstUnsentSeqNumFile);
     }
 
     private void closeFile(RandomAccessFile file) throws IOException {
@@ -234,6 +247,7 @@ public void deleteFiles() throws IOException {
         deleteFile(msgFileName);
         deleteFile(seqNumFileName);
         deleteFile(sessionFileName);
+        deleteFile(firstUnsentSeqNumFileName);
     }
 
     private void deleteFile(String fileName) throws IOException {
@@ -295,6 +309,26 @@ public void incrNextTargetMsgSeqNum() throws IOException {
         storeSequenceNumbers();
     }
 
+    /*
+     * (non-Javadoc)
+     * @see quickfix.MessageStore#getFirstUnsentSenderMsgSeqNum()
+     */
+    @Override
+    public int getFirstUnsentSenderMsgSeqNum() throws IOException {
+        return cache.getFirstUnsentSenderMsgSeqNum();
+    }
+
+    /*
+     * (non-Javadoc)
+     * @see quickfix.MessageStore#setFirstUnsentSenderMsgSeqNum(int)
+     */
+    @Override
+    public void setFirstUnsentSenderMsgSeqNum(int seqNum) throws IOException {
+        cache.setFirstUnsentSenderMsgSeqNum(seqNum);
+        firstUnsentSeqNumFile.seek(0);
+        firstUnsentSeqNumFile.writeUTF("" + seqNum);
+    }
+
     /*
      * (non-Javadoc)
      * @see quickfix.MessageStore#get(int, int, java.util.Collection)

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

@@ -259,6 +259,9 @@ public Session create(SessionID sessionID, SessionSettings settings) throws Conf
             final int maxScheduledWriteRequests = getSetting(settings, sessionID, Session.SETTING_MAX_SCHEDULED_WRITE_REQUESTS, 0);
             session.setMaxScheduledWriteRequests(maxScheduledWriteRequests);
 
+            final boolean keepUnsentMessagesOnReset = getSetting(settings, sessionID, Session.SETTING_KEEP_UNSENT_MESSAGES_ON_RESET, false);
+            session.setKeepUnsentMessagesOnReset(keepUnsentMessagesOnReset);
+
             //
             // Session registration and creation callback is done here instead of in
             // session constructor to eliminate the possibility of other threads

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

@@ -61,6 +61,7 @@ public class FileStore implements MessageStore, Closeable {
     private final String senderSeqNumFileName;
     private final String targetSeqNumFileName;
     private final String sessionFileName;
+    private final String firstUnsentSeqNumFileName;
     private final boolean syncWrites;
     private final int maxCachedMsgs;
     private RandomAccessFile messageFileReader;
@@ -69,6 +70,7 @@ public class FileStore implements MessageStore, Closeable {
     private FileOutputStream headerFileOutputStream;
     private RandomAccessFile senderSequenceNumberFile;
     private RandomAccessFile targetSequenceNumberFile;
+    private RandomAccessFile firstUnsentSeqNumFile;
 
     FileStore(String path, SessionID sessionID, boolean syncWrites, int maxCachedMsgs)
             throws IOException {
@@ -86,6 +88,7 @@ public class FileStore implements MessageStore, Closeable {
         senderSeqNumFileName = prefix + "senderseqnums";
         targetSeqNumFileName = prefix + "targetseqnums";
         sessionFileName = prefix + "session";
+        firstUnsentSeqNumFileName = prefix + "firstunsentseqnum";
 
         final File directory = new File(msgFileName).getParentFile();
         directory.mkdirs();
@@ -105,6 +108,7 @@ void initialize(boolean deleteFiles) throws IOException {
         messageFileReader = new RandomAccessFile(msgFileName, READ_OPTION);
         senderSequenceNumberFile = new RandomAccessFile(senderSeqNumFileName, mode);
         targetSequenceNumberFile = new RandomAccessFile(targetSeqNumFileName, mode);
+        firstUnsentSeqNumFile = new RandomAccessFile(firstUnsentSeqNumFileName, mode);
 
         initializeCache();
     }
@@ -170,6 +174,12 @@ private void initializeSequenceNumbers() throws IOException {
             final String s = targetSequenceNumberFile.readUTF();
             cache.setNextTargetMsgSeqNum(Integer.parseInt(s));
         }
+
+        firstUnsentSeqNumFile.seek(0);
+        if (firstUnsentSeqNumFile.length() > 0) {
+            final String s = firstUnsentSeqNumFile.readUTF();
+            cache.setFirstUnsentSenderMsgSeqNum(Integer.parseInt(s));
+        }
     }
 
     private void initializeMessageIndex() throws IOException {
@@ -216,6 +226,7 @@ public void close() throws IOException {
         close(messageFileReader);
         close(senderSequenceNumberFile);
         close(targetSequenceNumberFile);
+        close(firstUnsentSeqNumFile);
     }
 
     private static void close(Closeable closeable) throws IOException {
@@ -231,6 +242,7 @@ public void closeAndDeleteFiles() throws IOException {
         deleteFile(senderSeqNumFileName);
         deleteFile(targetSeqNumFileName);
         deleteFile(sessionFileName);
+        deleteFile(firstUnsentSeqNumFileName);
     }
 
     private void deleteFile(String fileName) throws IOException {
@@ -292,6 +304,24 @@ public void incrNextTargetMsgSeqNum() throws IOException {
         storeTargetSequenceNumber();
     }
 
+    /* (non-Javadoc)
+     * @see quickfix.MessageStore#getFirstUnsentSenderMsgSeqNum()
+     */
+    @Override
+    public int getFirstUnsentSenderMsgSeqNum() throws IOException {
+        return cache.getFirstUnsentSenderMsgSeqNum();
+    }
+
+    /* (non-Javadoc)
+     * @see quickfix.MessageStore#setFirstUnsentSenderMsgSeqNum(int)
+     */
+    @Override
+    public void setFirstUnsentSenderMsgSeqNum(int seqNum) throws IOException {
+        cache.setFirstUnsentSenderMsgSeqNum(seqNum);
+        firstUnsentSeqNumFile.seek(0);
+        firstUnsentSeqNumFile.writeUTF("" + seqNum);
+    }
+
     /* (non-Javadoc)
      * @see quickfix.MessageStore#get(int, int, java.util.Collection)
      */

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

@@ -63,6 +63,14 @@ public class JdbcSetting {
      */
     public static final String SETTING_JDBC_STORE_SESSIONS_TABLE_NAME = "JdbcStoreSessionsTableName";
 
+    /**
+     * Defines the table name for the table that keeps the first unsent sequence
+     * number. Default is "unsent_seqnums". This table is optional. It is only
+     * needed for {@link Session#SETTING_KEEP_UNSENT_MESSAGES_ON_RESET}.
+     * If you use a different name, you must set up your database accordingly.
+     */
+    public static final String SETTING_JDBC_STORE_UNSENT_SEQNUMS_TABLE_NAME = "JdbcStoreUnsentSeqNumsTableName";
+
     /**
      * The JNDI name used to lookup a DataSource for the JDBC plugins.
      */