Session implements Closeable { private void next(Message message, boolean isProcessingQueuedMessages) throws FieldNotFound, RejectLogon, IncorrectDataFormat, IncorrectTagValue, UnsupportedMessageType, IOException, InvalidMessage { if (message == EventHandlingStrategy.END_OF_STREAM) { disconnect(ENCOUNTERED_END_OF_STREAM, false); return; } final Header header = message.getHeader(); final String msgType = header.getString(MsgType.FIELD); if (!header.isSetField(MsgSeqNum.FIELD)) { generateLogout("Received message without MsgSeqNum"); disconnect("Received message without MsgSeqNum: " + getMessageToLog(message), true); return; } final String sessionBeginString = sessionID.getBeginString(); try { final String beginString = header.getString(BeginString.FIELD); if (!beginString.equals(sessionBeginString)) { throw new UnsupportedVersion("Message version '" + beginString + "' does not match the session version '" + sessionBeginString + "'"); } if (MsgType.LOGON.equals(msgType)) { if (sessionID.isFIXT()) { targetDefaultApplVerID.set(new ApplVerID(message .getString(DefaultApplVerID.FIELD))); } if (message.isSetField(HeartBtInt.FIELD)) { if (message.getInt(HeartBtInt.FIELD) < 0) { throw new RejectLogon("HeartBtInt must not be negative"); } } } if (validateIncomingMessage && dataDictionaryProvider != null) { final DataDictionary sessionDataDictionary = dataDictionaryProvider .getSessionDataDictionary(beginString); final ApplVerID applVerID = header.isSetField(ApplVerID.FIELD) ? new ApplVerID( header.getString(ApplVerID.FIELD)) : targetDefaultApplVerID.get(); final DataDictionary applicationDataDictionary = MessageUtils .isAdminMessage(msgType) ? dataDictionaryProvider .getSessionDataDictionary(beginString) : dataDictionaryProvider .getApplicationDataDictionary(applVerID); try { DataDictionary.validate(message, sessionDataDictionary, applicationDataDictionary); } catch (final IncorrectTagValue e) { if (rejectInvalidMessage) { throw e; } else { getLog().onErrorEvent("Warn: incoming message with " + e + ": " + getMessageToLog(message)); } } catch (final FieldException e) { if (message.isSetField(e.getField())) { if (rejectInvalidMessage) { throw e; } else { getLog().onErrorEvent( "Warn: incoming message with incorrect field: " + message.getField(e.getField()) + ": " + getMessageToLog(message)); } } else { if (rejectInvalidMessage) { throw e; } else { getLog().onErrorEvent( "Warn: incoming message with missing field: " + e.getField() + ": " + e.getMessage() + ": " + getMessageToLog(message)); } } } catch (final FieldNotFound e) { if (rejectInvalidMessage) { throw e; } else { getLog().onErrorEvent("Warn: incoming " + e + ": " + getMessageToLog(message)); } } } switch (msgType) { case MsgType.LOGON: nextLogon(message); break; case MsgType.HEARTBEAT: nextHeartBeat(message); break; case MsgType.TEST_REQUEST: nextTestRequest(message); break; case MsgType.SEQUENCE_RESET: nextSequenceReset(message); break; case MsgType.LOGOUT: nextLogout(message); break; case MsgType.RESEND_REQUEST: nextResendRequest(message); break; case MsgType.REJECT: nextReject(message); break; default: if (!verify(message)) { return; } state.incrNextTargetMsgSeqNum(); break; } } catch (final FieldException | IncorrectDataFormat | IncorrectTagValue e) { if (logErrorAndDisconnectIfRequired(e, message)) { return; } handleExceptionAndRejectMessage(msgType, message, e); } catch (final FieldNotFound e) { if (logErrorAndDisconnectIfRequired(e, message)) { return; } if (sessionBeginString.compareTo(FixVersions.BEGINSTRING_FIX42) >= 0 && message.isApp()) { generateBusinessReject(message, BusinessRejectReason.CONDITIONALLY_REQUIRED_FIELD_MISSING, e.field); } else { if (MsgType.LOGON.equals(msgType)) { getLog().onErrorEvent("Required field missing from logon"); disconnect("Required field missing from logon", true); } else { generateReject(message, SessionRejectReason.REQUIRED_TAG_MISSING, e.field); } } } catch (final InvalidMessage e) { if (rejectGarbledMessage) { getLog().onErrorEvent("Processing garbled message: " + e.getMessage()); generateReject(message, "Message failed basic validity check"); } else { getLog().onErrorEvent("Skipping invalid message: " + e + ": " + getMessageToLog(message)); if (resetOrDisconnectIfRequired(message)) { return; } } } catch (final RejectLogon e) { final String rejectMessage = e.getMessage() != null ? (": " + e) : ""; getLog().onErrorEvent("Logon rejected" + rejectMessage); if (e.isLogoutBeforeDisconnect()) { if (e.getSessionStatus() > -1) { generateLogout(e.getMessage(), new SessionStatus(e.getSessionStatus())); } else { generateLogout(e.getMessage()); } } if (getExpectedTargetNum() == header.getInt(MsgSeqNum.FIELD)) { state.incrNextTargetMsgSeqNum(); } disconnect("Logon rejected: " + e, true); } catch (final UnsupportedMessageType e) { if (logErrorAndDisconnectIfRequired(e, message)) { return; } if (sessionBeginString.compareTo(FixVersions.BEGINSTRING_FIX42) >= 0) { generateBusinessReject(message, BusinessRejectReason.UNSUPPORTED_MESSAGE_TYPE, 0); } else { generateReject(message, "Unsupported message type"); } } catch (final UnsupportedVersion e) { if (logErrorAndDisconnectIfRequired(e, message)) { return; } if (MsgType.LOGOUT.equals(msgType)) { nextLogout(message); } else { generateLogout("Incorrect BeginString: " + e.getMessage()); state.incrNextTargetMsgSeqNum(); disconnect("Incorrect BeginString: " + e, true); } } catch (final IOException e) { LogUtil.logThrowable(sessionID, "Error processing message: " + getMessageToLog(message), e); if (resetOrDisconnectIfRequired(message)) { return; } } catch (Throwable t) { if (rejectMessageOnUnhandledException) { getLog().onErrorEvent("Rejecting message: " + t + ": " + getMessageToLog(message)); if (resetOrDisconnectIfRequired(message)) { return; } if (!(MessageUtils.isAdminMessage(msgType)) && (sessionBeginString.compareTo(FixVersions.BEGINSTRING_FIX42) >= 0)) { generateBusinessReject(message, BusinessRejectReason.APPLICATION_NOT_AVAILABLE, 0); } else { if (MsgType.LOGON.equals(msgType)) { disconnect("Problem processing Logon message", true); } else { generateReject(message, SessionRejectReason.OTHER, 0); } } } else { throw new RuntimeError(t); } } if (!isProcessingQueuedMessages) { nextQueued(); if (isLoggedOn()) { next(); } } } Session(Application application, MessageStoreFactory messageStoreFactory, SessionID sessionID,
            DataDictionaryProvider dataDictionaryProvider, SessionSchedule sessionSchedule, LogFactory logFactory,
            MessageFactory messageFactory, int heartbeatInterval);   Session(Application application, MessageStoreFactory messageStoreFactory, SessionID sessionID,
            DataDictionaryProvider dataDictionaryProvider, SessionSchedule sessionSchedule,
            LogFactory logFactory, MessageFactory messageFactory, int heartbeatInterval,
            boolean checkLatency, int maxLatency, UtcTimestampPrecision timestampPrecision,
            boolean resetOnLogon, boolean resetOnLogout, boolean resetOnDisconnect,
            boolean refreshOnLogon, boolean checkCompID,
            boolean redundantResentRequestsAllowed, boolean persistMessages,
            boolean useClosedRangeForResend, double testRequestDelayMultiplier,
            DefaultApplVerID senderDefaultApplVerID, boolean validateSequenceNumbers,
            int[] logonIntervals, boolean resetOnError, boolean disconnectOnError,
            boolean disableHeartBeatCheck, boolean rejectGarbledMessage, boolean rejectInvalidMessage,
            boolean rejectMessageOnUnhandledException, boolean requiresOrigSendingTime,
            boolean forceResendWhenCorruptedStore, Set<InetAddress> allowedRemoteAddresses,
            boolean validateIncomingMessage, int resendRequestChunkSize,
            boolean enableNextExpectedMsgSeqNum, boolean enableLastMsgSeqNumProcessed,
            boolean validateChecksum, List<StringField> logonTags, double heartBeatTimeoutMultiplier); MessageFactory getMessageFactory(); void setResponder(Responder responder); Responder getResponder(); boolean hasResponder(); String getRemoteAddress(); static boolean sendToTarget(Message message); static boolean sendToTarget(Message message, String qualifier); static boolean sendToTarget(Message message, String senderCompID, String targetCompID); static boolean sendToTarget(Message message, String senderCompID, String targetCompID,
            String qualifier); static boolean sendToTarget(Message message, SessionID sessionID); static Session lookupSession(SessionID sessionID); void logon(); void logout(); void logout(String reason); boolean isEnabled(); boolean sentLogon(); boolean receivedLogon(); boolean sentLogout(); boolean receivedLogout(); boolean isLoggedOn(); void reset(); void setNextSenderMsgSeqNum(int num); void setNextTargetMsgSeqNum(int num); int getExpectedSenderNum(); int getExpectedTargetNum(); Log getLog(); MessageStore getStore(); void next(Message message); void generateLogout(); void next(); void generateHeartbeat(); void generateTestRequest(String id); void disconnect(String reason, boolean logError); boolean send(Message message); DataDictionary getDataDictionary(); DataDictionaryProvider getDataDictionaryProvider(); SessionID getSessionID(); boolean isSessionTime(); static boolean doesSessionExist(SessionID sessionID); static int numSessions(); void setLogonTimeout(int seconds); void setLogoutTimeout(int seconds); void setHeartBeatInterval(int heartbeatInterval); boolean getCheckCompID(); int getLogonTimeout(); int getLogoutTimeout(); boolean getRedundantResentRequestsAllowed(); boolean getRefreshOnLogon(); boolean getResetOnDisconnect(); boolean getResetOnLogout(); boolean isLogonAlreadySent(); boolean isLogonReceived(); boolean isLogonSendNeeded(); boolean isLogonSent(); boolean isLogonTimedOut(); boolean isLogoutReceived(); boolean isLogoutSent(); boolean isLogoutTimedOut(); boolean isValidateChecksum(); boolean isRejectGarbledMessage(); boolean isUsingDataDictionary(); Date getStartTime(); double getTestRequestDelayMultiplier(); @Override String toString(); void addStateListener(SessionStateListener listener); void removeStateListener(SessionStateListener listener); ApplVerID getSenderDefaultApplicationVersionID(); ApplVerID getTargetDefaultApplicationVersionID(); void setTargetDefaultApplicationVersionID(ApplVerID applVerID); int getMaxScheduledWriteRequests(); void setMaxScheduledWriteRequests(int maxScheduledWriteRequests); void setIgnoreHeartBeatFailure(boolean ignoreHeartBeatFailure); void setRejectGarbledMessage(boolean rejectGarbledMessage); void setValidateChecksum(
            final boolean validateChecksum); void setRejectInvalidMessage(boolean rejectInvalidMessage); void setRejectMessageOnUnhandledException(boolean rejectMessageOnUnhandledException); void setRequiresOrigSendingTime(boolean requiresOrigSendingTime); void setForceResendWhenCorruptedStore(boolean forceResendWhenCorruptedStore); boolean isAllowedForSession(InetAddress remoteInetAddress); @Override void close(); static final String SETTING_HEARTBTINT; static final String SETTING_CHECK_LATENCY; static final String SETTING_CHECK_COMP_ID; static final String SETTING_MAX_LATENCY; static final String SETTING_TEST_REQUEST_DELAY_MULTIPLIER; static final String SETTING_HEARTBEAT_TIMEOUT_MULTIPLIER; static final String SETTING_NON_STOP_SESSION; static final String SETTING_START_DAY; static final String SETTING_END_DAY; static final String SETTING_TIMEZONE; static final String SETTING_START_TIME; static final String SETTING_END_TIME; static final String SETTING_WEEKDAYS; static final String SETTING_USE_DATA_DICTIONARY; static final String SETTING_DATA_DICTIONARY; static final String SETTING_TRANSPORT_DATA_DICTIONARY; static final String SETTING_APP_DATA_DICTIONARY; static final String SETTING_VALIDATE_FIELDS_OUT_OF_ORDER; static final String SETTING_VALIDATE_UNORDERED_GROUP_FIELDS; static final String SETTING_VALIDATE_FIELDS_HAVE_VALUES; static final String SETTING_VALIDATE_INCOMING_MESSAGE; static final String SETTING_LOGON_TIMEOUT; static final String SETTING_LOGOUT_TIMEOUT; static final String SETTING_LOGON_TAG; static final String SETTING_RESET_ON_LOGOUT; static final String SETTING_VALIDATE_SEQUENCE_NUMBERS; static final String SETTING_RESET_ON_DISCONNECT; static final String SETTING_RESET_ON_ERROR; static final String SETTING_DISCONNECT_ON_ERROR; static final String SETTING_TIMESTAMP_PRECISION; static final String SETTING_VALIDATE_USER_DEFINED_FIELDS; static final String SETTING_RESET_ON_LOGON; static final String SETTING_DESCRIPTION; static final String SETTING_REFRESH_ON_LOGON; static final String SETTING_SEND_REDUNDANT_RESEND_REQUEST; static final String SETTING_PERSIST_MESSAGES; static final String SETTING_USE_CLOSED_RESEND_INTERVAL; static final String SETTING_ALLOW_UNKNOWN_MSG_FIELDS; static final String SETTING_DEFAULT_APPL_VER_ID; static final String SETTING_DISABLE_HEART_BEAT_CHECK; static final String SETTING_ENABLE_LAST_MSG_SEQ_NUM_PROCESSED; static final String SETTING_ENABLE_NEXT_EXPECTED_MSG_SEQ_NUM; static final String SETTING_REJECT_GARBLED_MESSAGE; static final String SETTING_REJECT_INVALID_MESSAGE; static final String SETTING_REJECT_MESSAGE_ON_UNHANDLED_EXCEPTION; static final String SETTING_REQUIRES_ORIG_SENDING_TIME; static final String SETTING_FORCE_RESEND_WHEN_CORRUPTED_STORE; static final String SETTING_ALLOWED_REMOTE_ADDRESSES; static final String SETTING_RESEND_REQUEST_CHUNK_SIZE; static final String SETTING_MAX_SCHEDULED_WRITE_REQUESTS; static final String SETTING_VALIDATE_CHECKSUM; static final int DEFAULT_MAX_LATENCY; static final int DEFAULT_RESEND_RANGE_CHUNK_SIZE; static final double DEFAULT_TEST_REQUEST_DELAY_MULTIPLIER; static final double DEFAULT_HEARTBEAT_TIMEOUT_MULTIPLIER; }
@Test public void testStartOfInitiatorInsideOfSessionTime() throws Exception { final LocalDateTime now = LocalDateTime.now(); ZoneOffset offset = ZoneOffset.systemDefault().getRules().getOffset(now); final MockSystemTimeSource systemTimeSource = new MockSystemTimeSource( now.toInstant(offset).toEpochMilli()); SystemTime.setTimeSource(systemTimeSource); final SessionID sessionID = new SessionID( FixVersions.BEGINSTRING_FIX44, "SENDER", "TARGET"); final SessionSettings settings = SessionSettingsTest.setUpSession(null); settings.setString("StartTime", UtcTimeOnlyConverter.convert(now.toLocalTime().minus(2000L, ChronoUnit.MILLIS), UtcTimestampPrecision.SECONDS)); settings.setString("EndTime", UtcTimeOnlyConverter.convert(now.toLocalTime().plus(3600000L, ChronoUnit.MILLIS), UtcTimestampPrecision.SECONDS)); settings.setString("TimeZone", TimeZone.getDefault().getID()); setupFileStoreForQFJ357(sessionID, settings); final UnitTestApplication application = new UnitTestApplication(); try (Session session = setUpFileStoreSession(application, true, new UnitTestResponder(), settings, sessionID)) { final SessionState state = getSessionState(session); assertEquals(1, state.getNextSenderMsgSeqNum()); assertEquals(1, state.getNextTargetMsgSeqNum()); session.next(); systemTimeSource.increment(1000); session.next(); systemTimeSource.increment(1000); session.next(); assertEquals(1, application.toAdminMessages.size()); assertEquals(MsgType.LOGON, application.toAdminMessages.get(0) .getHeader().getString(MsgType.FIELD)); assertEquals(0, application.sessionResets); assertEquals(2, state.getNextSenderMsgSeqNum()); assertEquals(1, state.getNextTargetMsgSeqNum()); } }