From 82903a8e275ae5e22b501bad42a0d5ef30010025 Mon Sep 17 00:00:00 2001 From: aymen Date: Wed, 30 Sep 2026 14:55:48 +0200 Subject: [PATCH 1/4] Add KeepUnsentMessagesOnReset option Before this change, some messages were lost. This happened when the app sent a message while the session was not logged on, for example outside the session time. The message was saved in the store, and then the scheduled reset deleted it. With KeepUnsentMessagesOnReset=Y, the message store keeps the sequence number of the first message that was not sent. When the session is reset by its schedule, the unsent application messages are copied to the new store with new sequence numbers. After the next logon, they are sent again when the counterparty asks for them. The value is saved in the store, so it works after a restart. It is supported by MemoryStore, FileStore and CachedFileStore. The default is N. So the reset works the same as before. Refs #463, #986 --- .../doc/usermanual/usage/configuration.md | 1 + .../main/java/quickfix/CachedFileStore.java | 34 +++ .../java/quickfix/DefaultSessionFactory.java | 3 + .../src/main/java/quickfix/FileStore.java | 30 ++ .../src/main/java/quickfix/MemoryStore.java | 12 + .../src/main/java/quickfix/MessageStore.java | 26 ++ .../src/main/java/quickfix/Session.java | 131 ++++++++- .../quickfix/AbstractMessageStoreTest.java | 24 ++ .../quickfix/DefaultSessionFactoryTest.java | 15 + .../src/test/java/quickfix/SessionTest.java | 257 ++++++++++++++++++ 10 files changed, 530 insertions(+), 3 deletions(-) diff --git a/quickfixj-core/src/main/doc/usermanual/usage/configuration.md b/quickfixj-core/src/main/doc/usermanual/usage/configuration.md index 7e227c67d6..5c41dea885 100644 --- a/quickfixj-core/src/main/doc/usermanual/usage/configuration.md +++ b/quickfixj-core/src/main/doc/usermanual/usage/configuration.md @@ -277,6 +277,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`
`N` | `N` | | `ForceResendWhenCorruptedStore` | Fill in heartbeats on resend when reading from message store fails. | `Y`
`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` or `CachedFileStore`. Messages are still lost with `ResetOnLogon`, `ResetOnLogout`, `ResetOnDisconnect`, a Logon with `ResetSeqNumFlag=Y` or a call to `Session.reset()`. | `Y`
`N` | `N` | | `LogMessageWhenSessionNotFound` | Log the entire message when the corresponding session can not be found. Otherwise only the SessionID is logged. | `Y`
`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`
`N` | `N` | diff --git a/quickfixj-core/src/main/java/quickfix/CachedFileStore.java b/quickfixj-core/src/main/java/quickfix/CachedFileStore.java index 12ba5f23a0..06b0ab0e27 100644 --- a/quickfixj-core/src/main/java/quickfix/CachedFileStore.java +++ b/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) diff --git a/quickfixj-core/src/main/java/quickfix/DefaultSessionFactory.java b/quickfixj-core/src/main/java/quickfix/DefaultSessionFactory.java index 59b8dd112a..cab4099197 100644 --- a/quickfixj-core/src/main/java/quickfix/DefaultSessionFactory.java +++ b/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 diff --git a/quickfixj-core/src/main/java/quickfix/FileStore.java b/quickfixj-core/src/main/java/quickfix/FileStore.java index 1d2309eb8d..043bc6b74c 100644 --- a/quickfixj-core/src/main/java/quickfix/FileStore.java +++ b/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) */ diff --git a/quickfixj-core/src/main/java/quickfix/MemoryStore.java b/quickfixj-core/src/main/java/quickfix/MemoryStore.java index 485611cb4e..cdc8f4f9ea 100644 --- a/quickfixj-core/src/main/java/quickfix/MemoryStore.java +++ b/quickfixj-core/src/main/java/quickfix/MemoryStore.java @@ -36,6 +36,7 @@ public class MemoryStore implements MessageStore { private final HashMap messages = new HashMap<>(); private int nextSenderMsgSeqNum; private int nextTargetMsgSeqNum; + private int firstUnsentSenderMsgSeqNum; private SessionID sessionID; private Calendar creationTimeCalendar = SystemTime.getUtcCalendar(); @@ -97,6 +98,7 @@ public void incrNextTargetMsgSeqNum() throws IOException { public void reset() throws IOException { setNextSenderMsgSeqNum(1); setNextTargetMsgSeqNum(1); + firstUnsentSenderMsgSeqNum = 0; messages.clear(); creationTimeCalendar = SystemTime.getUtcCalendar(); } @@ -113,6 +115,16 @@ public void setNextTargetMsgSeqNum(int next) throws IOException { nextTargetMsgSeqNum = next; } + @Override + public int getFirstUnsentSenderMsgSeqNum() { + return firstUnsentSenderMsgSeqNum; + } + + @Override + public void setFirstUnsentSenderMsgSeqNum(int seqNum) throws IOException { + firstUnsentSenderMsgSeqNum = seqNum; + } + public void refresh() throws IOException { // IOException is declared to maintain strict compatibility with QF JNI final String text = "memory store does not support refresh!"; diff --git a/quickfixj-core/src/main/java/quickfix/MessageStore.java b/quickfixj-core/src/main/java/quickfix/MessageStore.java index 756c72dad0..71255e4171 100644 --- a/quickfixj-core/src/main/java/quickfix/MessageStore.java +++ b/quickfixj-core/src/main/java/quickfix/MessageStore.java @@ -91,6 +91,32 @@ public interface MessageStore { */ void reset() throws IOException; + /** + * Get the sequence number of the first outgoing message that was stored + * but not sent yet. It is used to keep these messages when the session + * is reset by its schedule. + * + * @return the sequence number, 0 if all messages were sent, or -1 if + * this store does not support it. + * @throws IOException IO error + * @see Session#SETTING_KEEP_UNSENT_MESSAGES_ON_RESET + */ + default int getFirstUnsentSenderMsgSeqNum() throws IOException { + return -1; + } + + /** + * Set the sequence number of the first outgoing message that was stored + * but not sent yet. Use 0 when all messages were sent. The value is set + * back to 0 by {@link #reset()}. Stores that do not support it do nothing. + * + * @param seqNum the sequence number, or 0 + * @throws IOException IO error + * @see #getFirstUnsentSenderMsgSeqNum() + */ + default void setFirstUnsentSenderMsgSeqNum(int seqNum) throws IOException { + } + /** * Refresh session state from a shared state storage (e.g. database, * file, ...). Refresh will not work for message stores without shared state diff --git a/quickfixj-core/src/main/java/quickfix/Session.java b/quickfixj-core/src/main/java/quickfix/Session.java index 514de05c76..97561d6edd 100644 --- a/quickfixj-core/src/main/java/quickfix/Session.java +++ b/quickfixj-core/src/main/java/quickfix/Session.java @@ -376,6 +376,16 @@ public class Session implements Closeable { public static final String SETTING_FORCE_RESEND_WHEN_CORRUPTED_STORE = "ForceResendWhenCorruptedStore"; + /** + * Session setting to keep the application messages that were stored but not sent + * when the session is reset by its schedule. Without it, these messages are lost. + * The kept messages get new sequence numbers in the new session. They are sent + * again (with PossDupFlag=Y) when the counterparty asks for them after the next logon. + * The message store must support it (MemoryStore, FileStore and CachedFileStore do). + * Valid values are "Y" or "N". Default is "N". + */ + public static final String SETTING_KEEP_UNSENT_MESSAGES_ON_RESET = "KeepUnsentMessagesOnReset"; + public static final String SETTING_ALLOWED_REMOTE_ADDRESSES = "AllowedRemoteAddresses"; /** @@ -448,6 +458,7 @@ public class Session implements Closeable { private boolean rejectMessageOnUnhandledException = false; private boolean requiresOrigSendingTime = false; private boolean forceResendWhenCorruptedStore = false; + private boolean keepUnsentMessagesOnReset = false; private boolean enableNextExpectedMsgSeqNum = false; private boolean enableLastMsgSeqNumProcessed = false; private boolean validateChecksum = true; @@ -599,7 +610,13 @@ public class Session implements Closeable { getLog().onEvent("Session " + sessionID + " schedule is " + sessionSchedule); try { - resetIfSessionNotCurrent(sessionID, SystemTime.currentTimeMillis()); + if (sessionSchedule != null && messageStore.getFirstUnsentSenderMsgSeqNum() > 0) { + // The store has unsent messages. KeepUnsentMessagesOnReset is not set yet here, + // so the reset is done later by next() or before the logon. + getLog().onEvent("Message store has unsent messages; reset check is done later"); + } else { + resetIfSessionNotCurrent(sessionID, SystemTime.currentTimeMillis()); + } } catch (final IOException e) { LogUtil.logThrowable(getLog(), "error during session construction", e); } @@ -1975,8 +1992,8 @@ public void next() throws IOException { if ((now - lastSessionTimeCheck) >= 1000L) { lastSessionTimeCheck = now; if (!isSessionTime()) { - if (state.isResetNeeded() && !state.isResetStatePending()) { - reset(); // only reset if seq nums are != 1 and not isResetStatePending is set + if (isScheduledResetNeeded() && !state.isResetStatePending()) { + scheduledReset(); // only reset if seq nums are != 1 and not isResetStatePending is set } else if (state.isLogoutTimedOut()) { disconnect("Timed out waiting for logout response", true); } @@ -2361,6 +2378,10 @@ private void nextLogon(Message logon) throws FieldNotFound, RejectLogon, Incorre stateListener.onLogon(sessionID); lastSessionLogon = SystemTime.currentTimeMillis(); logonAttempts = 0; + // messages not sent yet are now sent on resend request from the counterparty + if (state.getMessageStore().getFirstUnsentSenderMsgSeqNum() > 0) { + state.getMessageStore().setFirstUnsentSenderMsgSeqNum(0); + } } } @@ -2731,6 +2752,9 @@ private boolean sendRaw(Message message, int num) { } catch (final Throwable t) { logApplicationException("toApp()", t); } + if (keepUnsentMessagesOnReset && num == 0 && !isLoggedOn()) { + markFirstUnsentMessage(header.getInt(MsgSeqNum.FIELD)); + } messageString = message.toString(); persist(message.getHeader(), messageString, num); if (isLoggedOn()) { @@ -3119,6 +3143,32 @@ public void setForceResendWhenCorruptedStore(boolean forceResendWhenCorruptedSto this.forceResendWhenCorruptedStore = forceResendWhenCorruptedStore; } + /** + * @param keepUnsentMessagesOnReset keep the unsent application messages when the session + * is reset by its schedule + * @see #SETTING_KEEP_UNSENT_MESSAGES_ON_RESET + */ + public void setKeepUnsentMessagesOnReset(boolean keepUnsentMessagesOnReset) { + this.keepUnsentMessagesOnReset = keepUnsentMessagesOnReset; + if (keepUnsentMessagesOnReset) { + try { + if (!persistMessages) { + getLog().onWarnEvent(SETTING_KEEP_UNSENT_MESSAGES_ON_RESET + + " has no effect because PersistMessages=N"); + } else if (state.getMessageStore().getFirstUnsentSenderMsgSeqNum() < 0) { + getLog().onWarnEvent(SETTING_KEEP_UNSENT_MESSAGES_ON_RESET + + " has no effect because the message store does not support it"); + } + } catch (final IOException e) { + logThrowable(getLog(), "Error reading from MessageStore", e); + } + } + } + + public boolean isKeepUnsentMessagesOnReset() { + return keepUnsentMessagesOnReset; + } + public boolean isAllowedForSession(InetAddress remoteInetAddress) { return allowedRemoteAddresses == null || allowedRemoteAddresses.isEmpty() || allowedRemoteAddresses.contains(remoteInetAddress); @@ -3149,10 +3199,85 @@ private void closeIfCloseable(Object resource) throws IOException { private void resetIfSessionNotCurrent(SessionID sessionID, long time) throws IOException { if (!isCurrentSession(time)) { getLog().onEvent("Session state is not current; resetting " + sessionID); + scheduledReset(); + } + } + + private boolean isScheduledResetNeeded() throws IOException { + if (keepUnsentMessagesOnReset && state.getNextTargetMsgSeqNum() == 1 + && state.getMessageStore().getFirstUnsentSenderMsgSeqNum() == 1) { + // the store only has unsent messages, a reset would do nothing new + return false; + } + return state.isResetNeeded(); + } + + /** + * Reset done by the session schedule. With KeepUnsentMessagesOnReset=Y and when + * not logged on, the application messages that were not sent are kept. + */ + private void scheduledReset() throws IOException { + if (keepUnsentMessagesOnReset && !isLoggedOn()) { + resetStateKeepingUnsentMessages(); + } else { reset(); } } + private void resetStateKeepingUnsentMessages() throws IOException { + state.lockSenderMsgSeqNum(); + try { + final List unsentMessages = getUnsentApplicationMessages(); + resetState(); + for (final Message message : unsentMessages) { + final Message.Header header = message.getHeader(); + final int msgSeqNum = getExpectedSenderNum(); + header.setInt(MsgSeqNum.FIELD, msgSeqNum); + insertSendingTime(header); + header.removeField(PossDupFlag.FIELD); + header.removeField(OrigSendingTime.FIELD); + markFirstUnsentMessage(msgSeqNum); + persist(header, message.toString(), 0); + } + if (!unsentMessages.isEmpty()) { + getLog().onEvent("Kept " + unsentMessages.size() + " unsent message(s) after reset"); + } + } catch (final FieldNotFound e) { + throw new IOException(e.getMessage(), e); + } finally { + state.unlockSenderMsgSeqNum(); + } + } + + private List getUnsentApplicationMessages() throws IOException { + final List unsentMessages = new ArrayList<>(); + final int firstUnsent = state.getMessageStore().getFirstUnsentSenderMsgSeqNum(); + if (firstUnsent <= 0) { + return unsentMessages; + } + final List messages = new ArrayList<>(); + state.get(firstUnsent, getExpectedSenderNum() - 1, messages); + for (final String messageData : messages) { + try { + final Message message = parseMessage(messageData); + if (!message.isAdmin()) { + unsentMessages.add(message); + } + } catch (final InvalidMessage e) { + getLog().onErrorEvent("Unsent message is lost, it can not be parsed (" + e.getMessage() + + "): " + messageData); + } + } + return unsentMessages; + } + + private void markFirstUnsentMessage(int msgSeqNum) throws IOException { + final MessageStore messageStore = state.getMessageStore(); + if (messageStore.getFirstUnsentSenderMsgSeqNum() == 0) { + messageStore.setFirstUnsentSenderMsgSeqNum(msgSeqNum); + } + } + private String getMessageToLog(final Message message) { return (message.toRawString() != null ? message.toRawString() : message.toString()); } diff --git a/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java b/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java index 40632bc471..997dc667cf 100644 --- a/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java +++ b/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java @@ -94,6 +94,30 @@ public void testMessageStoreSequenceNumbers() throws Exception { assertEquals("wrong value", 1, store.getNextTargetMsgSeqNum()); } + public void testFirstUnsentSenderMsgSeqNum() throws Exception { + if (!testEnabled) { + return; + } + if (!(store instanceof MemoryStore || store instanceof FileStore || store instanceof CachedFileStore)) { + // this store does not support it + assertEquals("wrong value", -1, store.getFirstUnsentSenderMsgSeqNum()); + return; + } + + store.reset(); + assertEquals("wrong value", 0, store.getFirstUnsentSenderMsgSeqNum()); + + store.setFirstUnsentSenderMsgSeqNum(5); + assertEquals("wrong value", 5, store.getFirstUnsentSenderMsgSeqNum()); + + // the value must be read again from the storage + store.refresh(); + assertEquals("wrong value", 5, store.getFirstUnsentSenderMsgSeqNum()); + + store.reset(); + assertEquals("wrong value", 0, store.getFirstUnsentSenderMsgSeqNum()); + } + protected SessionID getSessionID() { return sessionID; } diff --git a/quickfixj-core/src/test/java/quickfix/DefaultSessionFactoryTest.java b/quickfixj-core/src/test/java/quickfix/DefaultSessionFactoryTest.java index d3b6d3cdbf..a5e45f4b50 100644 --- a/quickfixj-core/src/test/java/quickfix/DefaultSessionFactoryTest.java +++ b/quickfixj-core/src/test/java/quickfix/DefaultSessionFactoryTest.java @@ -78,6 +78,21 @@ public void testAllowUnknownEnumValuesDefaultsToFalse() throws Exception { } } + @Test + public void testKeepUnsentMessagesOnResetSetting() throws Exception { + settings.setString(sessionID, Session.SETTING_KEEP_UNSENT_MESSAGES_ON_RESET, "Y"); + try (Session session = factory.create(sessionID, settings)) { + assertTrue(session.isKeepUnsentMessagesOnReset()); + } + } + + @Test + public void testKeepUnsentMessagesOnResetDefaultsToFalse() throws Exception { + try (Session session = factory.create(sessionID, settings)) { + assertFalse(session.isKeepUnsentMessagesOnReset()); + } + } + @Test public void testFixTMinimalSettings() throws Exception { sessionID = new SessionID(FixVersions.BEGINSTRING_FIXT11, "SENDER", "TARGET"); diff --git a/quickfixj-core/src/test/java/quickfix/SessionTest.java b/quickfixj-core/src/test/java/quickfix/SessionTest.java index 490f2ae456..04e9f6b66b 100644 --- a/quickfixj-core/src/test/java/quickfix/SessionTest.java +++ b/quickfixj-core/src/test/java/quickfix/SessionTest.java @@ -45,6 +45,7 @@ import java.io.FileOutputStream; import java.io.IOException; import java.lang.reflect.Field; +import java.nio.file.Files; import java.time.Duration; import java.time.Instant; import java.time.LocalDateTime; @@ -3468,6 +3469,262 @@ public void testDisabledSessionIsResetBySchedule() throws Exception { } } + /** + * https://github.com/quickfix-j/quickfixj/issues/463 + * https://github.com/quickfix-j/quickfixj/issues/986 + * Default behaviour: a message sent outside of the session time is lost + * when the session is reset by its schedule. + */ + @Test + public void testUnsentMessageIsLostOnScheduledResetByDefault() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, 3600000L, 7200000L)) { + final SessionState state = getSessionState(session); + assertFalse(session.isKeepUnsentMessagesOnReset()); + // messages of the previous session + session.setNextTargetMsgSeqNum(3); + + assertFalse(session.send(new News(new Headline("Unsent")))); + assertEquals(0, session.getStore().getFirstUnsentSenderMsgSeqNum()); + + session.next(); + + assertEquals(1, state.getNextSenderMsgSeqNum()); + assertEquals(1, state.getNextTargetMsgSeqNum()); + assertEquals(0, getStoredMessages(session).size()); + } + } + + /** + * With KeepUnsentMessagesOnReset=Y the unsent application messages are kept + * with new sequence numbers when the session is reset by its schedule. + * Admin messages are not kept. + */ + @Test + public void testUnsentMessageIsKeptOnScheduledReset() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + final MockSystemTimeSource systemTimeSource = setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + final SessionState state = getSessionState(session); + // messages of the previous session + session.setNextSenderMsgSeqNum(5); + session.setNextTargetMsgSeqNum(3); + + assertFalse(session.send(new News(new Headline("Unsent")))); + assertEquals(5, session.getStore().getFirstUnsentSenderMsgSeqNum()); + // an admin message in the unsent range is not kept + final Heartbeat heartbeat = new Heartbeat(); + setUpHeader(session.getSessionID(), heartbeat, false, 6); + session.getStore().set(6, heartbeat.toString()); + session.getStore().incrNextSenderMsgSeqNum(); + + session.next(); + + assertEquals(1, application.sessionResets); + assertEquals(2, state.getNextSenderMsgSeqNum()); + assertEquals(1, state.getNextTargetMsgSeqNum()); + assertEquals(1, session.getStore().getFirstUnsentSenderMsgSeqNum()); + final List stored = getStoredMessages(session); + assertEquals(1, stored.size()); + final Message kept = new Message(stored.get(0)); + assertEquals(News.MSGTYPE, kept.getHeader().getString(MsgType.FIELD)); + assertEquals("Unsent", kept.getString(Headline.FIELD)); + assertEquals(1, kept.getHeader().getInt(MsgSeqNum.FIELD)); + assertFalse(kept.getHeader().isSetField(PossDupFlag.FIELD)); + assertEquals(1, application.toAppMessages.size()); + + // the store has only unsent messages: no new reset outside of the session time + systemTimeSource.increment(1000L); + session.next(); + assertEquals(1, application.sessionResets); + assertEquals(2, state.getNextSenderMsgSeqNum()); + } + } + + /** + * With KeepUnsentMessagesOnReset=Y the kept message survives the reset done when the + * session time starts, and it is resent when the counterparty asks for it after logon. + */ + @Test + public void testUnsentMessageIsResentAfterLogon() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + final MockSystemTimeSource systemTimeSource = setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + final RecordingResponder responder = new RecordingResponder(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + session.setResponder(responder); + final SessionState state = getSessionState(session); + + assertFalse(session.send(new News(new Headline("Unsent")))); + assertTrue(responder.sentMessages.isEmpty()); + + // start of the session time: the store is not current and is reset + systemTimeSource.increment(3601000L); + session.next(); + assertEquals(1, application.sessionResets); + assertEquals(2, state.getNextSenderMsgSeqNum()); + + logonTo(session); + assertTrue(session.isLoggedOn()); + assertEquals(0, session.getStore().getFirstUnsentSenderMsgSeqNum()); + final Message logonResponse = new Message(responder.getLastMessage()); + assertEquals(MsgType.LOGON, logonResponse.getHeader().getString(MsgType.FIELD)); + assertEquals(2, logonResponse.getHeader().getInt(MsgSeqNum.FIELD)); + + responder.sentMessages.clear(); + processMessage(session, createResendRequest(2, 1)); + + // the News message, then a SequenceReset for the Logon response + final Message resent = new Message(responder.sentMessages.get(0)); + assertEquals(News.MSGTYPE, resent.getHeader().getString(MsgType.FIELD)); + assertEquals("Unsent", resent.getString(Headline.FIELD)); + assertEquals(1, resent.getHeader().getInt(MsgSeqNum.FIELD)); + assertTrue(resent.getHeader().getBoolean(PossDupFlag.FIELD)); + } + } + + /** + * With KeepUnsentMessagesOnReset=Y the messages that were sent while logged on + * are not kept, so they are not sent two times. + */ + @Test + public void testSentMessageIsNotKeptOnScheduledReset() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + final MockSystemTimeSource systemTimeSource = setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + final RecordingResponder responder = new RecordingResponder(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, -3600000L, 3600000L)) { + session.setKeepUnsentMessagesOnReset(true); + session.setResponder(responder); + + logonTo(session); + assertTrue(session.isLoggedOn()); + assertTrue(session.send(new News(new Headline("Sent")))); + assertEquals(0, session.getStore().getFirstUnsentSenderMsgSeqNum()); + session.disconnect("test", false); + + // end of the session time + systemTimeSource.increment(3601000L); + assertFalse(session.send(new News(new Headline("Unsent")))); + assertEquals(3, session.getStore().getFirstUnsentSenderMsgSeqNum()); + + session.next(); + + final List stored = getStoredMessages(session); + assertEquals(1, stored.size()); + final Message kept = new Message(stored.get(0)); + assertEquals("Unsent", kept.getString(Headline.FIELD)); + assertEquals(1, kept.getHeader().getInt(MsgSeqNum.FIELD)); + } + } + + /** + * With KeepUnsentMessagesOnReset=Y and a FileStore the unsent messages are + * not lost when the application is restarted. + */ + @Test + public void testUnsentMessageIsKeptAfterRestart() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + final MockSystemTimeSource systemTimeSource = setUpMockSystemTime(now); + final SessionSettings storeSettings = new SessionSettings(); + final File storeDir = Files.createTempDirectory("qfj-unsent").toFile(); + storeSettings.setString(FileStoreFactory.SETTING_FILE_STORE_PATH, storeDir.getAbsolutePath()); + final FileStoreFactory storeFactory = new FileStoreFactory(storeSettings); + try { + try (Session session = createScheduledAcceptorSession(new UnitTestApplication(), storeFactory, now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + assertFalse(session.send(new News(new Headline("Unsent")))); + assertEquals(1, session.getStore().getFirstUnsentSenderMsgSeqNum()); + } + + // restart at the start of the session time + systemTimeSource.increment(3601000L); + final UnitTestApplication application = new UnitTestApplication(); + try (Session session = createScheduledAcceptorSession(application, storeFactory, now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + // the reset is not done in the constructor + assertEquals(2, getSessionState(session).getNextSenderMsgSeqNum()); + + session.next(); + + assertEquals(1, application.sessionResets); + final List stored = getStoredMessages(session); + assertEquals(1, stored.size()); + assertEquals("Unsent", new Message(stored.get(0)).getString(Headline.FIELD)); + ((FileStore) session.getStore()).closeAndDeleteFiles(); + } + } finally { + final File[] files = storeDir.listFiles(); + if (files != null) { + for (File file : files) { + file.delete(); + } + } + storeDir.delete(); + } + } + + private MockSystemTimeSource setUpMockSystemTime(LocalDateTime now) { + final ZoneOffset offset = ZoneOffset.systemDefault().getRules().getOffset(now); + final MockSystemTimeSource systemTimeSource = new MockSystemTimeSource(now.toInstant(offset).toEpochMilli()); + SystemTime.setTimeSource(systemTimeSource); + return systemTimeSource; + } + + private Session createScheduledAcceptorSession(UnitTestApplication application, MessageStoreFactory storeFactory, + LocalDateTime now, long startOffsetMillis, long endOffsetMillis) throws Exception { + final SessionID sessionID = new SessionID(FixVersions.BEGINSTRING_FIX44, "SENDER", "TARGET"); + final SessionSettings settings = SessionSettingsTest.setUpSession(null); + settings.setString("StartTime", UtcTimeOnlyConverter.convert(now.toLocalTime().plus(startOffsetMillis, ChronoUnit.MILLIS), UtcTimestampPrecision.SECONDS)); + settings.setString("EndTime", UtcTimeOnlyConverter.convert(now.toLocalTime().plus(endOffsetMillis, ChronoUnit.MILLIS), UtcTimestampPrecision.SECONDS)); + settings.setString("TimeZone", TimeZone.getDefault().getID()); + + final Session session = new SessionFactoryTestSupport.Builder() + .setSessionId(sessionID) + .setApplication(application) + .setMessageStoreFactory(storeFactory) + .setSessionSchedule(new DefaultSessionSchedule(settings, sessionID)) + .setIsInitiator(false) + .setPersistMessages(true) + .build(); + session.addStateListener(application); + return session; + } + + private List getStoredMessages(Session session) throws IOException { + final List messages = new ArrayList<>(); + session.getStore().get(1, 100, messages); + return messages; + } + + private static class RecordingResponder implements Responder { + final List sentMessages = new ArrayList<>(); + + @Override + public boolean send(String data) { + sentMessages.add(data); + return true; + } + + String getLastMessage() { + return sentMessages.get(sentMessages.size() - 1); + } + + @Override + public String getRemoteAddress() { + return null; + } + + @Override + public void disconnect() { + } + } + /** * https://github.com/quickfix-j/quickfixj/issues/646 * Verify that resend operations abort when send() returns false. From 965a23329846d5e4b6c076ddd8f27e3cead7edfc Mon Sep 17 00:00:00 2001 From: aymen Date: Thu, 15 Oct 2026 02:32:53 +0200 Subject: [PATCH 2/4] feat(store): support KeepUnsentMessagesOnReset in JdbcStore JdbcStore can now keep the first unsent sequence number. The value is saved in a new table, "unsent_seqnums". The name can be changed with JdbcStoreUnsentSeqNumsTableName. The table is optional. Without it, JdbcStore works as before and KeepUnsentMessagesOnReset has no effect. The SQL scripts for HSQLDB, MySQL, MSSQL, Oracle and PostgreSQL create the new table. Refs #463, #986 --- .../doc/usermanual/usage/configuration.md | 3 +- .../src/main/java/quickfix/JdbcSetting.java | 8 ++ .../src/main/java/quickfix/JdbcStore.java | 99 +++++++++++++++++++ .../src/main/java/quickfix/JdbcUtil.java | 9 ++ .../sql/hsqldb/unsent_seqnums_table.sql | 15 +++ .../config/sql/mssql/quickfix_database.sql | 14 +++ .../resources/config/sql/mysql/create_all.sql | 19 ++++ .../main/resources/config/sql/mysql/mysql.sql | 1 + .../config/sql/mysql/unsent_seqnums_table.sql | 17 ++++ .../sql/oracle/unsent_seqnums_table.sql | 13 +++ .../config/sql/postgresql/postgresql.sql | 1 + .../sql/postgresql/unsent_seqnums_table.sql | 13 +++ .../quickfix/AbstractMessageStoreTest.java | 3 +- .../java/quickfix/JdbcStoreLegacyTest.java | 3 + .../src/test/java/quickfix/JdbcStoreTest.java | 22 +++++ 15 files changed, 238 insertions(+), 2 deletions(-) create mode 100644 quickfixj-core/src/main/resources/config/sql/hsqldb/unsent_seqnums_table.sql create mode 100644 quickfixj-core/src/main/resources/config/sql/mysql/unsent_seqnums_table.sql create mode 100644 quickfixj-core/src/main/resources/config/sql/oracle/unsent_seqnums_table.sql create mode 100644 quickfixj-core/src/main/resources/config/sql/postgresql/unsent_seqnums_table.sql diff --git a/quickfixj-core/src/main/doc/usermanual/usage/configuration.md b/quickfixj-core/src/main/doc/usermanual/usage/configuration.md index 5c41dea885..1ac36c4a13 100644 --- a/quickfixj-core/src/main/doc/usermanual/usage/configuration.md +++ b/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`
`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,7 +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`
`N` | `N` | | `ForceResendWhenCorruptedStore` | Fill in heartbeats on resend when reading from message store fails. | `Y`
`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` or `CachedFileStore`. Messages are still lost with `ResetOnLogon`, `ResetOnLogout`, `ResetOnDisconnect`, a Logon with `ResetSeqNumFlag=Y` or a call to `Session.reset()`. | `Y`
`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`
`N` | `N` | | `LogMessageWhenSessionNotFound` | Log the entire message when the corresponding session can not be found. Otherwise only the SessionID is logged. | `Y`
`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`
`N` | `N` | diff --git a/quickfixj-core/src/main/java/quickfix/JdbcSetting.java b/quickfixj-core/src/main/java/quickfix/JdbcSetting.java index 7653ee7f59..7c01fdbc06 100644 --- a/quickfixj-core/src/main/java/quickfix/JdbcSetting.java +++ b/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. */ diff --git a/quickfixj-core/src/main/java/quickfix/JdbcStore.java b/quickfixj-core/src/main/java/quickfix/JdbcStore.java index 9954db7c71..4318eb6cee 100644 --- a/quickfixj-core/src/main/java/quickfix/JdbcStore.java +++ b/quickfixj-core/src/main/java/quickfix/JdbcStore.java @@ -33,6 +33,7 @@ class JdbcStore implements MessageStore { private final static String DEFAULT_SESSION_TABLE_NAME = "sessions"; private final static String DEFAULT_MESSAGE_TABLE_NAME = "messages"; + private final static String DEFAULT_UNSENT_SEQNUMS_TABLE_NAME = "unsent_seqnums"; private final MemoryStore cache = new MemoryStore(); private final boolean extendedSessionIdSupported; @@ -40,8 +41,10 @@ class JdbcStore implements MessageStore { private final SessionID sessionID; private final String sessionTableName; private final String messageTableName; + private final String unsentSeqNumsTableName; private final String defaultSessionIdPropertyValue; private final boolean persistMessages; + private final boolean firstUnsentSeqNumSupported; private String SQL_UPDATE_INCOMING_SEQNUM; private String SQL_UPDATE_OUTGOING_SEQNUM; @@ -52,12 +55,16 @@ class JdbcStore implements MessageStore { private String SQL_GET_MESSAGES; private String SQL_UPDATE_SESSION; private String SQL_DELETE_MESSAGES; + private String SQL_GET_FIRST_UNSENT_SEQNUM; + private String SQL_UPDATE_FIRST_UNSENT_SEQNUM; + private String SQL_INSERT_FIRST_UNSENT_SEQNUM; public JdbcStore(SessionSettings settings, SessionID sessionID, DataSource ds) throws Exception { this.sessionID = sessionID; sessionTableName = getSessionTableName(settings, sessionID); messageTableName = getMessageTableName(settings, sessionID); + unsentSeqNumsTableName = getUnsentSeqNumsTableName(settings, sessionID); if (settings.isSetting(sessionID, SETTING_JDBC_SESSION_ID_DEFAULT_PROPERTY_VALUE)) { defaultSessionIdPropertyValue = settings.getString(sessionID, @@ -76,6 +83,11 @@ public JdbcStore(SessionSettings settings, SessionID sessionID, DataSource ds) t extendedSessionIdSupported = JdbcUtil.determineSessionIdSupport(dataSource, sessionTableName); + // The unsent sequence number table is optional. Without it, the store + // does not support KeepUnsentMessagesOnReset. + firstUnsentSeqNumSupported = JdbcUtil.determineFirstUnsentSeqNumSupport(dataSource, + unsentSeqNumsTableName); + setSqlStrings(); loadCache(); @@ -97,6 +109,14 @@ public static String getMessageTableName(SessionSettings settings, SessionID ses } } + public static String getUnsentSeqNumsTableName(SessionSettings settings, SessionID sessionID) throws ConfigError { + if (settings.isSetting(sessionID, SETTING_JDBC_STORE_UNSENT_SEQNUMS_TABLE_NAME)) { + return settings.getString(sessionID, SETTING_JDBC_STORE_UNSENT_SEQNUMS_TABLE_NAME); + } else { + return DEFAULT_UNSENT_SEQNUMS_TABLE_NAME; + } + } + public static String getUpdateIncomingSequenceNumberSql(String sessionTableName, String idWhereClause) { return "UPDATE " + sessionTableName + " SET incoming_seqnum=? WHERE " + idWhereClause; } @@ -133,6 +153,18 @@ public static String getDeleteMessagesSql(String messageTableName, String idWher return "DELETE FROM " + messageTableName + " WHERE " + idWhereClause; } + public static String getFirstUnsentSequenceNumSql(String unsentSeqNumsTableName, String idWhereClause) { + return "SELECT first_unsent_seqnum FROM " + unsentSeqNumsTableName + " WHERE " + idWhereClause; + } + + public static String getUpdateFirstUnsentSequenceNumSql(String unsentSeqNumsTableName, String idWhereClause) { + return "UPDATE " + unsentSeqNumsTableName + " SET first_unsent_seqnum=? WHERE " + idWhereClause; + } + + public static String getInsertFirstUnsentSequenceNumSql(String unsentSeqNumsTableName, String idColumns, String idPlaceholders) { + return "INSERT INTO " + unsentSeqNumsTableName + " (" + idColumns + ", first_unsent_seqnum) VALUES (" + idPlaceholders + ",?)"; + } + private void setSqlStrings() { String idWhereClause = JdbcUtil.getIDWhereClause(extendedSessionIdSupported); String idColumns = JdbcUtil.getIDColumns(extendedSessionIdSupported); @@ -147,13 +179,18 @@ private void setSqlStrings() { SQL_GET_MESSAGES = getMessagesSql(messageTableName, idWhereClause); SQL_UPDATE_SESSION = getUpdateSessionSql(sessionTableName, idWhereClause); SQL_DELETE_MESSAGES = getDeleteMessagesSql(messageTableName, idWhereClause); + SQL_GET_FIRST_UNSENT_SEQNUM = getFirstUnsentSequenceNumSql(unsentSeqNumsTableName, idWhereClause); + SQL_UPDATE_FIRST_UNSENT_SEQNUM = getUpdateFirstUnsentSequenceNumSql(unsentSeqNumsTableName, idWhereClause); + SQL_INSERT_FIRST_UNSENT_SEQNUM = getInsertFirstUnsentSequenceNumSql(unsentSeqNumsTableName, idColumns, idPlaceholders); } private void loadCache() throws SQLException, IOException { Connection connection = null; PreparedStatement query = null; PreparedStatement insert = null; + PreparedStatement unsentQuery = null; ResultSet rs = null; + ResultSet unsentRs = null; try { connection = dataSource.getConnection(); query = connection.prepareStatement(SQL_GET_SEQNUMS); @@ -171,7 +208,16 @@ private void loadCache() throws SQLException, IOException { insert.setInt(offset, cache.getNextSenderMsgSeqNum()); insert.execute(); } + if (firstUnsentSeqNumSupported) { + unsentQuery = connection.prepareStatement(SQL_GET_FIRST_UNSENT_SEQNUM); + setSessionIdParameters(unsentQuery, 1); + unsentRs = unsentQuery.executeQuery(); + // no row means that all messages were sent + cache.setFirstUnsentSenderMsgSeqNum(unsentRs.next() ? unsentRs.getInt(1) : 0); + } } finally { + JdbcUtil.close(sessionID, unsentRs); + JdbcUtil.close(sessionID, unsentQuery); JdbcUtil.close(sessionID, rs); JdbcUtil.close(sessionID, query); JdbcUtil.close(sessionID, insert); @@ -215,6 +261,7 @@ public void reset() throws IOException { Connection connection = null; PreparedStatement deleteMessages = null; PreparedStatement updateTime = null; + PreparedStatement updateFirstUnsent = null; try { connection = dataSource.getConnection(); if (persistMessages) { @@ -230,11 +277,19 @@ public void reset() throws IOException { updateTime.setInt(3, getNextSenderMsgSeqNum()); setSessionIdParameters(updateTime, 4); updateTime.execute(); + + if (firstUnsentSeqNumSupported) { + updateFirstUnsent = connection.prepareStatement(SQL_UPDATE_FIRST_UNSENT_SEQNUM); + updateFirstUnsent.setInt(1, 0); + setSessionIdParameters(updateFirstUnsent, 2); + updateFirstUnsent.execute(); + } } catch (SQLException e) { throw new IOException(e.getMessage(), e); } finally { JdbcUtil.close(sessionID, deleteMessages); JdbcUtil.close(sessionID, updateTime); + JdbcUtil.close(sessionID, updateFirstUnsent); JdbcUtil.close(sessionID, connection); } } @@ -307,6 +362,50 @@ public void setNextTargetMsgSeqNum(int next) throws IOException { storeSequenceNumber(SQL_UPDATE_INCOMING_SEQNUM, next); } + /* (non-Javadoc) + * @see quickfix.MessageStore#getFirstUnsentSenderMsgSeqNum() + */ + @Override + public int getFirstUnsentSenderMsgSeqNum() throws IOException { + if (!firstUnsentSeqNumSupported) { + return -1; + } + return cache.getFirstUnsentSenderMsgSeqNum(); + } + + /* (non-Javadoc) + * @see quickfix.MessageStore#setFirstUnsentSenderMsgSeqNum(int) + */ + @Override + public void setFirstUnsentSenderMsgSeqNum(int seqNum) throws IOException { + if (!firstUnsentSeqNumSupported) { + return; + } + cache.setFirstUnsentSenderMsgSeqNum(seqNum); + Connection connection = null; + PreparedStatement update = null; + PreparedStatement insert = null; + try { + connection = dataSource.getConnection(); + update = connection.prepareStatement(SQL_UPDATE_FIRST_UNSENT_SEQNUM); + update.setInt(1, seqNum); + setSessionIdParameters(update, 2); + if (update.executeUpdate() == 0) { + // first value for this session + insert = connection.prepareStatement(SQL_INSERT_FIRST_UNSENT_SEQNUM); + int offset = setSessionIdParameters(insert, 1); + insert.setInt(offset, seqNum); + insert.execute(); + } + } catch (SQLException e) { + throw new IOException(e.getMessage(), e); + } finally { + JdbcUtil.close(sessionID, update); + JdbcUtil.close(sessionID, insert); + JdbcUtil.close(sessionID, connection); + } + } + private void storeSequenceNumber(String sequenceUpdateSql, int sequence) throws IOException { Connection connection = null; PreparedStatement update = null; diff --git a/quickfixj-core/src/main/java/quickfix/JdbcUtil.java b/quickfixj-core/src/main/java/quickfix/JdbcUtil.java index c6ce2c2fda..a6014094fc 100644 --- a/quickfixj-core/src/main/java/quickfix/JdbcUtil.java +++ b/quickfixj-core/src/main/java/quickfix/JdbcUtil.java @@ -164,6 +164,15 @@ static boolean determineSessionIdSupport(DataSource dataSource, String tableName } } + static boolean determineFirstUnsentSeqNumSupport(DataSource dataSource, String tableName) throws SQLException { + try (Connection connection = dataSource.getConnection()) { + DatabaseMetaData metaData = connection.getMetaData(); + String columnName = "first_unsent_seqnum"; + return isColumn(metaData, tableName.toUpperCase(), columnName.toUpperCase()) + || isColumn(metaData, tableName, columnName); + } + } + private static boolean isColumn(DatabaseMetaData metaData, String tableName, String columnName) throws SQLException { try (ResultSet columns = metaData.getColumns(null, null, tableName, columnName)) { diff --git a/quickfixj-core/src/main/resources/config/sql/hsqldb/unsent_seqnums_table.sql b/quickfixj-core/src/main/resources/config/sql/hsqldb/unsent_seqnums_table.sql new file mode 100644 index 0000000000..ca874e1faa --- /dev/null +++ b/quickfixj-core/src/main/resources/config/sql/hsqldb/unsent_seqnums_table.sql @@ -0,0 +1,15 @@ +USE quickfix; +DROP TABLE unsent_seqnums IF EXISTS +CREATE TABLE unsent_seqnums ( + beginstring CHAR(8) NOT NULL, + sendercompid VARCHAR(64) NOT NULL, + sendersubid VARCHAR(64) NOT NULL, + senderlocid VARCHAR(64) NOT NULL, + targetcompid VARCHAR(64) NOT NULL, + targetsubid VARCHAR(64) NOT NULL, + targetlocid VARCHAR(64) NOT NULL, + session_qualifier VARCHAR(64) NOT NULL, + first_unsent_seqnum INT NOT NULL, + PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, + targetcompid, targetsubid, targetlocid, session_qualifier) +); diff --git a/quickfixj-core/src/main/resources/config/sql/mssql/quickfix_database.sql b/quickfixj-core/src/main/resources/config/sql/mssql/quickfix_database.sql index 4e59e37668..ad991cbe46 100644 --- a/quickfixj-core/src/main/resources/config/sql/mssql/quickfix_database.sql +++ b/quickfixj-core/src/main/resources/config/sql/mssql/quickfix_database.sql @@ -18,6 +18,20 @@ CREATE TABLE sessions ( targetcompid, targetsubid, targetlocid, session_qualifier) ); +CREATE TABLE unsent_seqnums ( + beginstring CHAR(8) NOT NULL, + sendercompid VARCHAR(64) NOT NULL, + sendersubid VARCHAR(64) NOT NULL, + senderlocid VARCHAR(64) NOT NULL, + targetcompid VARCHAR(64) NOT NULL, + targetsubid VARCHAR(64) NOT NULL, + targetlocid VARCHAR(64) NOT NULL, + session_qualifier VARCHAR(64) NOT NULL, + first_unsent_seqnum INT NOT NULL, + PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, + targetcompid, targetsubid, targetlocid, session_qualifier) +); + CREATE TABLE messages ( beginstring CHAR(8) NOT NULL, sendercompid VARCHAR(64) NOT NULL, diff --git a/quickfixj-core/src/main/resources/config/sql/mysql/create_all.sql b/quickfixj-core/src/main/resources/config/sql/mysql/create_all.sql index 99bf6e3cfd..c21f5ab542 100644 --- a/quickfixj-core/src/main/resources/config/sql/mysql/create_all.sql +++ b/quickfixj-core/src/main/resources/config/sql/mysql/create_all.sql @@ -23,6 +23,25 @@ PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, ); +USE quickfix; + +DROP TABLE IF EXISTS unsent_seqnums; + +CREATE TABLE unsent_seqnums ( +beginstring CHAR(8) NOT NULL, +sendercompid VARCHAR(64) NOT NULL, +sendersubid VARCHAR(64) NOT NULL, +senderlocid VARCHAR(64) NOT NULL, +targetcompid VARCHAR(64) NOT NULL, +targetsubid VARCHAR(64) NOT NULL, +targetlocid VARCHAR(64) NOT NULL, +session_qualifier VARCHAR(64) NOT NULL, +first_unsent_seqnum INT NOT NULL, +PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, + targetcompid, targetsubid, targetlocid, session_qualifier) +); + + USE quickfix; DROP TABLE IF EXISTS messages; diff --git a/quickfixj-core/src/main/resources/config/sql/mysql/mysql.sql b/quickfixj-core/src/main/resources/config/sql/mysql/mysql.sql index a130b5b78b..a9b2bb7cd2 100644 --- a/quickfixj-core/src/main/resources/config/sql/mysql/mysql.sql +++ b/quickfixj-core/src/main/resources/config/sql/mysql/mysql.sql @@ -1,5 +1,6 @@ source quickfix_database.sql; source sessions_table.sql; +source unsent_seqnums_table.sql; source messages_table.sql; source messages_log_table.sql; source event_log_table.sql; \ No newline at end of file diff --git a/quickfixj-core/src/main/resources/config/sql/mysql/unsent_seqnums_table.sql b/quickfixj-core/src/main/resources/config/sql/mysql/unsent_seqnums_table.sql new file mode 100644 index 0000000000..0d04fffb25 --- /dev/null +++ b/quickfixj-core/src/main/resources/config/sql/mysql/unsent_seqnums_table.sql @@ -0,0 +1,17 @@ +USE quickfix; + +DROP TABLE IF EXISTS unsent_seqnums; + +CREATE TABLE unsent_seqnums ( + beginstring CHAR(8) NOT NULL, + sendercompid VARCHAR(64) NOT NULL, + sendersubid VARCHAR(64) NOT NULL, + senderlocid VARCHAR(64) NOT NULL, + targetcompid VARCHAR(64) NOT NULL, + targetsubid VARCHAR(64) NOT NULL, + targetlocid VARCHAR(64) NOT NULL, + session_qualifier VARCHAR(64) NOT NULL, + first_unsent_seqnum INT NOT NULL, + PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, + targetcompid, targetsubid, targetlocid, session_qualifier) +); diff --git a/quickfixj-core/src/main/resources/config/sql/oracle/unsent_seqnums_table.sql b/quickfixj-core/src/main/resources/config/sql/oracle/unsent_seqnums_table.sql new file mode 100644 index 0000000000..5b638a5129 --- /dev/null +++ b/quickfixj-core/src/main/resources/config/sql/oracle/unsent_seqnums_table.sql @@ -0,0 +1,13 @@ +CREATE TABLE unsent_seqnums ( + beginstring VARCHAR2(8) NOT NULL, + sendercompid VARCHAR2(64) NOT NULL, + sendersubid VARCHAR2(64) NOT NULL, + senderlocid VARCHAR2(64) NOT NULL, + targetcompid VARCHAR2(64) NOT NULL, + targetsubid VARCHAR2(64) NOT NULL, + targetlocid VARCHAR2(64) NOT NULL, + session_qualifier VARCHAR2(64) NOT NULL, + first_unsent_seqnum INTEGER NOT NULL, + PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, + targetcompid, targetsubid, targetlocid, session_qualifier) +); diff --git a/quickfixj-core/src/main/resources/config/sql/postgresql/postgresql.sql b/quickfixj-core/src/main/resources/config/sql/postgresql/postgresql.sql index c54d0b9f12..54dbf0ab35 100644 --- a/quickfixj-core/src/main/resources/config/sql/postgresql/postgresql.sql +++ b/quickfixj-core/src/main/resources/config/sql/postgresql/postgresql.sql @@ -1,4 +1,5 @@ \i sessions_table.sql; +\i unsent_seqnums_table.sql; \i messages_table.sql; \i messages_log_table.sql; \i event_log_table.sql; \ No newline at end of file diff --git a/quickfixj-core/src/main/resources/config/sql/postgresql/unsent_seqnums_table.sql b/quickfixj-core/src/main/resources/config/sql/postgresql/unsent_seqnums_table.sql new file mode 100644 index 0000000000..8497f14b9c --- /dev/null +++ b/quickfixj-core/src/main/resources/config/sql/postgresql/unsent_seqnums_table.sql @@ -0,0 +1,13 @@ +CREATE TABLE unsent_seqnums ( + beginstring CHAR(8) NOT NULL, + sendercompid VARCHAR(64) NOT NULL, + sendersubid VARCHAR(64) NOT NULL, + senderlocid VARCHAR(64) NOT NULL, + targetcompid VARCHAR(64) NOT NULL, + targetsubid VARCHAR(64) NOT NULL, + targetlocid VARCHAR(64) NOT NULL, + session_qualifier VARCHAR(64) NOT NULL, + first_unsent_seqnum INTEGER NOT NULL, + PRIMARY KEY (beginstring, sendercompid, sendersubid, senderlocid, + targetcompid, targetsubid, targetlocid, session_qualifier) +); diff --git a/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java b/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java index 997dc667cf..f919ccf6aa 100644 --- a/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java +++ b/quickfixj-core/src/test/java/quickfix/AbstractMessageStoreTest.java @@ -98,7 +98,8 @@ public void testFirstUnsentSenderMsgSeqNum() throws Exception { if (!testEnabled) { return; } - if (!(store instanceof MemoryStore || store instanceof FileStore || store instanceof CachedFileStore)) { + if (!(store instanceof MemoryStore || store instanceof FileStore || store instanceof CachedFileStore + || store instanceof JdbcStore)) { // this store does not support it assertEquals("wrong value", -1, store.getFirstUnsentSenderMsgSeqNum()); return; diff --git a/quickfixj-core/src/test/java/quickfix/JdbcStoreLegacyTest.java b/quickfixj-core/src/test/java/quickfix/JdbcStoreLegacyTest.java index fc38ce7c5d..7b75defc4a 100644 --- a/quickfixj-core/src/test/java/quickfix/JdbcStoreLegacyTest.java +++ b/quickfixj-core/src/test/java/quickfix/JdbcStoreLegacyTest.java @@ -40,6 +40,9 @@ protected void initializeTableDefinitions(String sessionsTableName, String messa JdbcTestSupport.loadSQL(connection, "config/sql/hsqldb/sessions_table.sql", new JdbcTestSupport.HypersonicLegacyPreprocessor(sessionsTableName)); + JdbcTestSupport.loadSQL(connection, + "config/sql/hsqldb/unsent_seqnums_table.sql", + new JdbcTestSupport.HypersonicLegacyPreprocessor(null)); } finally { JdbcUtil.close(null, connection); } diff --git a/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java b/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java index e442f9881e..ecdde1f8aa 100644 --- a/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java +++ b/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java @@ -165,11 +165,33 @@ protected void initializeTableDefinitions(String sessionsTableName, String messa } loadSQL(connection, "config/sql/hsqldb/sessions_table.sql", new JdbcTestSupport.HypersonicPreprocessor(sessionsTableName)); + loadSQL(connection, "config/sql/hsqldb/unsent_seqnums_table.sql", + new JdbcTestSupport.HypersonicPreprocessor(null)); } finally { close(null, connection); } } + public void testFirstUnsentSenderMsgSeqNumWithoutTable() throws Exception { + Connection connection = null; + try { + connection = getTestDataSource().getConnection(); + dropTable(connection, "unsent_seqnums"); + } finally { + close(null, connection); + } + SessionSettings settings = new SessionSettings(); + settings.setString(SETTING_JDBC_DS_NAME, "TestDataSource"); + JdbcStore store = (JdbcStore) new JdbcStoreFactory(settings).create(getSessionID()); + + // the table is optional: the store works, but does not support it + assertEquals("wrong value", -1, store.getFirstUnsentSenderMsgSeqNum()); + store.setFirstUnsentSenderMsgSeqNum(5); + assertEquals("wrong value", -1, store.getFirstUnsentSenderMsgSeqNum()); + store.reset(); + assertEquals("wrong value", 1, store.getNextSenderMsgSeqNum()); + } + protected DataSource getTestDataSource() { return JdbcTestSupport.getTestDataSource(HSQL_DRIVER, HSQL_CONNECTION_URL, HSQL_USER, ""); } From d7444bf877fa4249cdceca4b0d4ccd62aca1d7c8 Mon Sep 17 00:00:00 2001 From: aymen Date: Thu, 15 Oct 2026 12:35:06 +0200 Subject: [PATCH 3/4] test(session): cover more KeepUnsentMessagesOnReset cases Add tests for several unsent messages, a scheduled reset without unsent messages, the warnings when the setting has no effect, and the update and custom table name in JdbcStore. --- .../src/test/java/quickfix/JdbcStoreTest.java | 48 +++++++++ .../src/test/java/quickfix/SessionTest.java | 102 ++++++++++++++++++ 2 files changed, 150 insertions(+) diff --git a/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java b/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java index ecdde1f8aa..1d85fddd74 100644 --- a/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java +++ b/quickfixj-core/src/test/java/quickfix/JdbcStoreTest.java @@ -33,6 +33,7 @@ import static quickfix.JdbcSetting.SETTING_JDBC_DS_NAME; import static quickfix.JdbcSetting.SETTING_JDBC_STORE_MESSAGES_TABLE_NAME; import static quickfix.JdbcSetting.SETTING_JDBC_STORE_SESSIONS_TABLE_NAME; +import static quickfix.JdbcSetting.SETTING_JDBC_STORE_UNSENT_SEQNUMS_TABLE_NAME; import static quickfix.JdbcTestSupport.HSQL_CONNECTION_URL; import static quickfix.JdbcTestSupport.HSQL_DRIVER; import static quickfix.JdbcTestSupport.HSQL_USER; @@ -172,6 +173,53 @@ protected void initializeTableDefinitions(String sessionsTableName, String messa } } + public void testFirstUnsentSenderMsgSeqNumIsUpdated() throws Exception { + JdbcStore store = (JdbcStore) getStore(); + store.reset(); + + store.setFirstUnsentSenderMsgSeqNum(5); + // the row exists now, so it is updated + store.setFirstUnsentSenderMsgSeqNum(7); + + // a new store reads the value from the database + JdbcStore newStore = (JdbcStore) createStore(); + assertEquals("wrong value", 7, newStore.getFirstUnsentSenderMsgSeqNum()); + + store.setFirstUnsentSenderMsgSeqNum(0); + newStore.refresh(); + assertEquals("wrong value", 0, newStore.getFirstUnsentSenderMsgSeqNum()); + } + + public void testFirstUnsentSenderMsgSeqNumWithCustomTableName() throws Exception { + Connection connection = null; + try { + connection = getTestDataSource().getConnection(); + dropTable(connection, "xunsent"); + JdbcTestSupport.execSQL(connection, "ALTER TABLE unsent_seqnums RENAME TO xunsent"); + } finally { + close(null, connection); + } + try { + SessionSettings settings = new SessionSettings(); + settings.setString(SETTING_JDBC_DS_NAME, "TestDataSource"); + settings.setString(SETTING_JDBC_STORE_UNSENT_SEQNUMS_TABLE_NAME, "xunsent"); + JdbcStore store = (JdbcStore) new JdbcStoreFactory(settings).create(getSessionID()); + store.reset(); + assertEquals("wrong value", 0, store.getFirstUnsentSenderMsgSeqNum()); + + store.setFirstUnsentSenderMsgSeqNum(5); + store.refresh(); + assertEquals("wrong value", 5, store.getFirstUnsentSenderMsgSeqNum()); + } finally { + try { + connection = getTestDataSource().getConnection(); + dropTable(connection, "xunsent"); + } finally { + close(null, connection); + } + } + } + public void testFirstUnsentSenderMsgSeqNumWithoutTable() throws Exception { Connection connection = null; try { diff --git a/quickfixj-core/src/test/java/quickfix/SessionTest.java b/quickfixj-core/src/test/java/quickfix/SessionTest.java index 04e9f6b66b..d329033e40 100644 --- a/quickfixj-core/src/test/java/quickfix/SessionTest.java +++ b/quickfixj-core/src/test/java/quickfix/SessionTest.java @@ -3676,6 +3676,108 @@ private MockSystemTimeSource setUpMockSystemTime(LocalDateTime now) { return systemTimeSource; } + /** + * With KeepUnsentMessagesOnReset=Y all the unsent messages are kept, in the same order. + */ + @Test + public void testSeveralUnsentMessagesAreKeptInOrderOnScheduledReset() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + final SessionState state = getSessionState(session); + // messages of the previous session + session.setNextSenderMsgSeqNum(5); + session.setNextTargetMsgSeqNum(3); + + assertFalse(session.send(new News(new Headline("Unsent1")))); + assertFalse(session.send(new News(new Headline("Unsent2")))); + assertFalse(session.send(new News(new Headline("Unsent3")))); + // the first unsent message does not change + assertEquals(5, session.getStore().getFirstUnsentSenderMsgSeqNum()); + + session.next(); + + assertEquals(1, application.sessionResets); + assertEquals(4, state.getNextSenderMsgSeqNum()); + assertEquals(1, session.getStore().getFirstUnsentSenderMsgSeqNum()); + final List stored = getStoredMessages(session); + assertEquals(3, stored.size()); + for (int i = 0; i < 3; i++) { + final Message kept = new Message(stored.get(i)); + assertEquals("Unsent" + (i + 1), kept.getString(Headline.FIELD)); + assertEquals(i + 1, kept.getHeader().getInt(MsgSeqNum.FIELD)); + } + } + } + + /** + * With KeepUnsentMessagesOnReset=Y and no unsent message, the scheduled reset + * works the same as before. + */ + @Test + public void testScheduledResetWithoutUnsentMessages() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + final SessionState state = getSessionState(session); + // messages of the previous session, all sent + for (int seqNum = 1; seqNum <= 4; seqNum++) { + final News news = new News(new Headline("Sent" + seqNum)); + setUpHeader(session.getSessionID(), news, false, seqNum); + session.getStore().set(seqNum, news.toString()); + } + session.setNextSenderMsgSeqNum(5); + session.setNextTargetMsgSeqNum(3); + assertEquals(4, getStoredMessages(session).size()); + assertEquals(0, session.getStore().getFirstUnsentSenderMsgSeqNum()); + + session.next(); + + assertEquals(1, application.sessionResets); + assertEquals(1, state.getNextSenderMsgSeqNum()); + assertEquals(1, state.getNextTargetMsgSeqNum()); + assertEquals(0, session.getStore().getFirstUnsentSenderMsgSeqNum()); + assertEquals(0, getStoredMessages(session).size()); + } + } + + /** + * KeepUnsentMessagesOnReset=Y logs a warning when it can not work. + */ + @Test + public void testKeepUnsentMessagesOnResetWarnsWhenItHasNoEffect() throws Exception { + final SessionID sessionID = new SessionID(FixVersions.BEGINSTRING_FIX44, "SENDER", "TARGET"); + final LogFactory mockLogFactory = mock(LogFactory.class); + final Log mockLog = mock(Log.class); + when(mockLogFactory.create(sessionID)).thenReturn(mockLog); + + try (Session session = new SessionFactoryTestSupport.Builder() + .setSessionId(sessionID) + .setLogFactory(mockLogFactory) + .setPersistMessages(false) + .build()) { + session.setKeepUnsentMessagesOnReset(true); + } + verify(mockLog).onWarnEvent(Session.SETTING_KEEP_UNSENT_MESSAGES_ON_RESET + + " has no effect because PersistMessages=N"); + + // NoopStore does not support it + try (Session session = new SessionFactoryTestSupport.Builder() + .setSessionId(sessionID) + .setLogFactory(mockLogFactory) + .setMessageStoreFactory(new NoopStoreFactory()) + .setPersistMessages(true) + .build()) { + session.setKeepUnsentMessagesOnReset(true); + } + verify(mockLog).onWarnEvent(Session.SETTING_KEEP_UNSENT_MESSAGES_ON_RESET + + " has no effect because the message store does not support it"); + } + private Session createScheduledAcceptorSession(UnitTestApplication application, MessageStoreFactory storeFactory, LocalDateTime now, long startOffsetMillis, long endOffsetMillis) throws Exception { final SessionID sessionID = new SessionID(FixVersions.BEGINSTRING_FIX44, "SENDER", "TARGET"); From aa3486ccda54a51554b67594ff42c676bbe7c385 Mon Sep 17 00:00:00 2001 From: Aymen Jamaaoui <5993228+aymenjamaaoui@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:42:50 +0200 Subject: [PATCH 4/4] test(session): check that unsent XMLnonFIX messages are kept --- .../src/test/java/quickfix/SessionTest.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/quickfixj-core/src/test/java/quickfix/SessionTest.java b/quickfixj-core/src/test/java/quickfix/SessionTest.java index d329033e40..d97b5e456d 100644 --- a/quickfixj-core/src/test/java/quickfix/SessionTest.java +++ b/quickfixj-core/src/test/java/quickfix/SessionTest.java @@ -3712,6 +3712,39 @@ public void testSeveralUnsentMessagesAreKeptInOrderOnScheduledReset() throws Exc } } + /** + * With KeepUnsentMessagesOnReset=Y an unsent XMLnonFIX message is kept. + * It is a session level message, but it is not an admin message for QFJ. + */ + @Test + public void testUnsentXmlNonFixMessageIsKeptOnScheduledReset() throws Exception { + final LocalDateTime now = LocalDateTime.now().truncatedTo(ChronoUnit.SECONDS); + setUpMockSystemTime(now); + final UnitTestApplication application = new UnitTestApplication(); + try (Session session = createScheduledAcceptorSession(application, new MemoryStoreFactory(), now, 3600000L, 7200000L)) { + session.setKeepUnsentMessagesOnReset(true); + final SessionState state = getSessionState(session); + // messages of the previous session + session.setNextSenderMsgSeqNum(5); + session.setNextTargetMsgSeqNum(3); + + final Message xmlNonFix = new Message(); + xmlNonFix.getHeader().setString(MsgType.FIELD, MsgType.XMLNON_FIX); + assertFalse(session.send(xmlNonFix)); + assertEquals(5, session.getStore().getFirstUnsentSenderMsgSeqNum()); + + session.next(); + + assertEquals(1, application.sessionResets); + assertEquals(2, state.getNextSenderMsgSeqNum()); + final List stored = getStoredMessages(session); + assertEquals(1, stored.size()); + final Message kept = new Message(stored.get(0)); + assertEquals(MsgType.XMLNON_FIX, kept.getHeader().getString(MsgType.FIELD)); + assertEquals(1, kept.getHeader().getInt(MsgSeqNum.FIELD)); + } + } + /** * With KeepUnsentMessagesOnReset=Y and no unsent message, the scheduled reset * works the same as before.