diff --git a/artio-core/src/main/java/uk/co/real_logic/artio/engine/framer/Framer.java b/artio-core/src/main/java/uk/co/real_logic/artio/engine/framer/Framer.java index 1c3b717932..46060fedcd 100644 --- a/artio-core/src/main/java/uk/co/real_logic/artio/engine/framer/Framer.java +++ b/artio-core/src/main/java/uk/co/real_logic/artio/engine/framer/Framer.java @@ -1523,6 +1523,7 @@ private void onFixConnectionOpen( sendRedundantResendRequests, enableLastMsgSeqNumProcessed, fixDictionary); + gatewaySession.libraryId(libraryId); gatewaySession.lastSequenceResetTime(sessionContext.lastSequenceResetTime()); gatewaySession.lastLogonTime(sessionContext.lastLogonTimeInNs()); library.addSession(gatewaySession); @@ -1833,8 +1834,7 @@ public Action onLibraryConnect( { existingLibrary.onHeartbeat(epochClock.time()); - return Pressure.apply(inboundPublication.saveControlNotification( - libraryId, initialAcceptedSessionOwner, existingLibrary.sessions(), NO_DISCONNECTED_SESSIONS)); + return saveControlNotification(libraryId, existingLibrary.sessions()) ? CONTINUE : ABORT; } if (soleLibraryMode) @@ -1851,17 +1851,10 @@ public Action onLibraryConnect( } // Send an empty control notification if you've never seen this library before - final LongHashSet disconnectedSessions = gatewaySessions.findDisconnectedSessions(libraryId); - if (Pressure.isBackPressured( - inboundPublication.saveControlNotification( - libraryId, initialAcceptedSessionOwner, Collections.emptyList(), disconnectedSessions))) + if (!saveControlNotification(libraryId, Collections.emptyList())) { return ABORT; } - else - { - gatewaySessions.removeDisconnectedSessions(disconnectedSessions); - } final List unitsOfWork = new ArrayList<>(); unitsOfWork.add(() -> @@ -1908,6 +1901,22 @@ public Action onLibraryConnect( return retryManager.firstAttempt(correlationId, new UnitOfWork(unitsOfWork)); } + private boolean saveControlNotification(final int libraryId, final List sessions) + { + final LongHashSet disconnectedSessions = gatewaySessions.findDisconnectedSessions(libraryId); + if (Pressure.isBackPressured( + inboundPublication.saveControlNotification( + libraryId, initialAcceptedSessionOwner, sessions, disconnectedSessions))) + { + return false; + } + else + { + gatewaySessions.removeDisconnectedSessions(disconnectedSessions); + return true; + } + } + private void soleLibraryModeBind() { if (shouldBind)