Skip to content

Commit ad0acfd

Browse files
cbusbeycbusbey
authored
Merge pull request #233 from cbusbey/resend_toapp_toadmin
calls toApp toAdmin on resend and seq reset
2 parents a72230f + 7f2fb2d commit ad0acfd

File tree

5 files changed

+192
-20
lines changed

5 files changed

+192
-20
lines changed

‎in_session.go‎

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -220,21 +220,31 @@ func (state inSession) resendMessages(session *session, beginSeqNo, endSeqNo int
220220
continue
221221
}
222222

223+
if !session.resend(msg) {
224+
nextSeqNum = sentMessageSeqNum + 1
225+
continue
226+
}
227+
223228
if seqNum != sentMessageSeqNum {
224-
state.generateSequenceReset(session, seqNum, sentMessageSeqNum)
229+
if err = state.generateSequenceReset(session, seqNum, sentMessageSeqNum); err != nil {
230+
return err
231+
}
225232
}
226233

227234
session.log.OnEventf("Resending Message: %v", sentMessageSeqNum)
228-
if err = session.resend(msg); err != nil {
229-
return
235+
if _, err := msg.Build(); err != nil {
236+
return err
230237
}
238+
session.sendBytes(msg.rawMessage)
231239

232240
seqNum = sentMessageSeqNum + 1
233241
nextSeqNum = seqNum
234242
}
235243

236244
if seqNum != nextSeqNum { // gapfill for catch-up
237-
state.generateSequenceReset(session, seqNum, nextSeqNum)
245+
if err = state.generateSequenceReset(session, seqNum, nextSeqNum); err != nil {
246+
return err
247+
}
238248
}
239249

240250
return
@@ -335,7 +345,7 @@ func (state inSession) doTargetTooLow(session *session, msg Message, rej targetT
335345
return state
336346
}
337347

338-
func (state *inSession) generateSequenceReset(session *session, beginSeqNo int, endSeqNo int) {
348+
func (state *inSession) generateSequenceReset(session *session, beginSeqNo int, endSeqNo int) (err error) {
339349
sequenceReset := NewMessage()
340350
session.fillDefaultHeader(sequenceReset)
341351

@@ -350,9 +360,15 @@ func (state *inSession) generateSequenceReset(session *session, beginSeqNo int,
350360
sequenceReset.Header.SetField(tagOrigSendingTime, origSendingTime)
351361
}
352362

353-
//FIXME error check?
354-
msgBytes, _ := sequenceReset.Build()
355-
session.sendBytes(msgBytes)
363+
session.application.ToAdmin(sequenceReset, session.sessionID)
356364

365+
msgBytes, err := sequenceReset.Build()
366+
if err != nil {
367+
return
368+
}
369+
370+
session.sendBytes(msgBytes)
357371
session.log.OnEventf("Sent SequenceReset TO: %v", endSeqNo)
372+
373+
return
358374
}

‎in_session_test.go‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,3 +133,112 @@ func (s *InSessionTestSuite) TestFIXMsgInTargetTooHigh() {
133133
stashedRawMsg, _ := stashedMsg.Build()
134134
s.Equal(string(rawMsg), string(stashedRawMsg))
135135
}
136+
137+
func (s *InSessionTestSuite) TestFIXMsgInResendRequestAllAdminExpectGapFill() {
138+
s.MockApp.On("ToAdmin")
139+
s.session.Timeout(s.session, internal.NeedHeartbeat)
140+
s.LastToAdminMessageSent()
141+
142+
s.session.Timeout(s.session, internal.NeedHeartbeat)
143+
s.LastToAdminMessageSent()
144+
145+
s.session.Timeout(s.session, internal.NeedHeartbeat)
146+
s.LastToAdminMessageSent()
147+
148+
s.MockApp.AssertNumberOfCalls(s.T(), "ToAdmin", 3)
149+
s.NextSenderMsgSeqNum(4)
150+
151+
s.MockApp.On("FromAdmin").Return(nil)
152+
s.MockApp.On("ToAdmin")
153+
s.fixMsgIn(s.session, s.ResendRequest(1))
154+
155+
s.MockApp.AssertExpectations(s.T())
156+
s.LastToAdminMessageSent()
157+
s.MessageType(enum.MsgType_SEQUENCE_RESET, s.MockApp.lastToAdmin)
158+
s.FieldEquals(tagMsgSeqNum, 1, s.MockApp.lastToAdmin.Header)
159+
s.FieldEquals(tagPossDupFlag, true, s.MockApp.lastToAdmin.Header)
160+
s.FieldEquals(tagNewSeqNo, 4, s.MockApp.lastToAdmin.Body)
161+
s.FieldEquals(tagGapFillFlag, true, s.MockApp.lastToAdmin.Body)
162+
163+
s.NextSenderMsgSeqNum(4)
164+
s.State(inSession{})
165+
}
166+
167+
func (s *InSessionTestSuite) TestFIXMsgInResendRequestAllAdminThenApp() {
168+
s.MockApp.On("ToAdmin")
169+
s.session.Timeout(s.session, internal.NeedHeartbeat)
170+
s.LastToAdminMessageSent()
171+
172+
s.session.Timeout(s.session, internal.NeedHeartbeat)
173+
s.LastToAdminMessageSent()
174+
175+
s.MockApp.On("ToApp").Return(nil)
176+
s.session.send(s.NewOrderSingle())
177+
s.LastToAppMessageSent()
178+
179+
s.MockApp.AssertNumberOfCalls(s.T(), "ToAdmin", 2)
180+
s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 1)
181+
s.NextSenderMsgSeqNum(4)
182+
183+
s.MockApp.On("FromAdmin").Return(nil)
184+
s.MockApp.On("ToAdmin")
185+
s.MockApp.On("ToApp").Return(nil)
186+
s.fixMsgIn(s.session, s.ResendRequest(1))
187+
188+
s.MockApp.AssertNumberOfCalls(s.T(), "ToAdmin", 3)
189+
s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 2)
190+
191+
s.LastToAdminMessageSent()
192+
s.MessageType(enum.MsgType_SEQUENCE_RESET, s.MockApp.lastToAdmin)
193+
s.FieldEquals(tagMsgSeqNum, 1, s.MockApp.lastToAdmin.Header)
194+
s.FieldEquals(tagPossDupFlag, true, s.MockApp.lastToAdmin.Header)
195+
s.FieldEquals(tagNewSeqNo, 3, s.MockApp.lastToAdmin.Body)
196+
s.FieldEquals(tagGapFillFlag, true, s.MockApp.lastToAdmin.Body)
197+
198+
s.LastToAppMessageSent()
199+
s.MessageType(enum.MsgType_ORDER_SINGLE, s.MockApp.lastToApp)
200+
s.FieldEquals(tagMsgSeqNum, 3, s.MockApp.lastToApp.Header)
201+
s.FieldEquals(tagPossDupFlag, true, s.MockApp.lastToApp.Header)
202+
203+
s.NextSenderMsgSeqNum(4)
204+
s.State(inSession{})
205+
}
206+
207+
func (s *InSessionTestSuite) TestFIXMsgInResendRequestDoNotSendApp() {
208+
s.MockApp.On("ToAdmin")
209+
s.session.Timeout(s.session, internal.NeedHeartbeat)
210+
s.LastToAdminMessageSent()
211+
212+
s.MockApp.On("ToApp").Return(nil)
213+
s.session.send(s.NewOrderSingle())
214+
s.LastToAppMessageSent()
215+
216+
s.session.Timeout(s.session, internal.NeedHeartbeat)
217+
s.LastToAdminMessageSent()
218+
219+
s.MockApp.AssertNumberOfCalls(s.T(), "ToAdmin", 2)
220+
s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 1)
221+
s.NextSenderMsgSeqNum(4)
222+
223+
//NOTE: a cheat here, need to reset mock
224+
s.MockApp = MockApp{}
225+
s.MockApp.On("FromAdmin").Return(nil)
226+
s.MockApp.On("ToApp").Return(ErrDoNotSend)
227+
s.MockApp.On("ToAdmin")
228+
s.fixMsgIn(s.session, s.ResendRequest(1))
229+
230+
s.MockApp.AssertNumberOfCalls(s.T(), "ToAdmin", 1)
231+
s.MockApp.AssertNumberOfCalls(s.T(), "ToApp", 1)
232+
233+
s.LastToAdminMessageSent()
234+
s.MessageType(enum.MsgType_SEQUENCE_RESET, s.MockApp.lastToAdmin)
235+
s.FieldEquals(tagMsgSeqNum, 1, s.MockApp.lastToAdmin.Header)
236+
s.FieldEquals(tagPossDupFlag, true, s.MockApp.lastToAdmin.Header)
237+
s.FieldEquals(tagNewSeqNo, 4, s.MockApp.lastToAdmin.Body)
238+
s.FieldEquals(tagGapFillFlag, true, s.MockApp.lastToAdmin.Body)
239+
240+
s.NoMessageSent()
241+
242+
s.NextSenderMsgSeqNum(4)
243+
s.State(inSession{})
244+
}

‎quickfix_test.go‎

Lines changed: 26 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ type KnowsFieldMap interface {
1717
Has(Tag) bool
1818
GetString(Tag) (string, MessageRejectError)
1919
GetInt(Tag) (int, MessageRejectError)
20+
GetField(Tag, FieldValueReader) MessageRejectError
2021
}
2122

2223
func (s *QuickFIXSuite) MessageType(msgType string, msg Message) {
@@ -35,6 +36,11 @@ func (s *QuickFIXSuite) FieldEquals(tag Tag, expectedValue interface{}, fieldMap
3536
val, err := fieldMap.GetInt(tag)
3637
s.Nil(err)
3738
s.Equal(expected, val)
39+
case bool:
40+
var val FIXBoolean
41+
err := fieldMap.GetField(tag, &val)
42+
s.Nil(err)
43+
s.Equal(expected, val.Bool())
3844
default:
3945
s.FailNow("Field type not handled")
4046
}
@@ -122,19 +128,34 @@ func (m *MessageFactory) buildMessage(msgType string) Message {
122128
}
123129

124130
func (m *MessageFactory) Logout() Message {
125-
return m.buildMessage("5")
131+
return m.buildMessage(enum.MsgType_LOGOUT)
126132
}
127133

128134
func (m *MessageFactory) NewOrderSingle() Message {
129-
return m.buildMessage("D")
135+
return m.buildMessage(enum.MsgType_ORDER_SINGLE)
130136
}
131137

132138
func (m *MessageFactory) Heartbeat() Message {
133-
return m.buildMessage("0")
139+
return m.buildMessage(enum.MsgType_HEARTBEAT)
134140
}
135141

136142
func (m *MessageFactory) Logon() Message {
137-
return m.buildMessage("A")
143+
return m.buildMessage(enum.MsgType_LOGON)
144+
}
145+
146+
func (m *MessageFactory) ResendRequest(beginSeqNo int) Message {
147+
msg := m.buildMessage(enum.MsgType_RESEND_REQUEST)
148+
msg.Body.SetField(tagBeginSeqNo, FIXInt(beginSeqNo))
149+
msg.Body.SetField(tagEndSeqNo, FIXInt(0))
150+
151+
return msg
152+
}
153+
154+
func (m *MessageFactory) SequenceReset(seqNo int) Message {
155+
msg := m.buildMessage(enum.MsgType_SEQUENCE_RESET)
156+
msg.Body.SetField(tagNewSeqNo, FIXInt(seqNo))
157+
158+
return msg
138159
}
139160

140161
type MockSessionReceiver struct {
@@ -216,7 +237,7 @@ func (s *SessionSuiteRig) Disconnected() {
216237

217238
func (s *SessionSuiteRig) NoMessageSent() {
218239
msg, _ := s.Receiver.LastMessage()
219-
s.Nil(msg, "no message should be sent")
240+
s.Nil(msg, "no message should be sent but got %s", msg)
220241
}
221242

222243
func (s *SessionSuiteRig) NoMessageQueued() {

‎resend_state_test.go‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,3 +95,34 @@ func (s *resendStateTestSuite) TestFixMsgIn() {
9595
s.State(inSession{})
9696
s.NextTargetMsgSeqNum(5)
9797
}
98+
99+
func (s *resendStateTestSuite) TestFixMsgInSequenceReset() {
100+
s.session.State = inSession{}
101+
102+
//in session expects seq number 1, send too high
103+
s.MessageFactory.SetNextSeqNum(3)
104+
s.MockApp.On("ToAdmin")
105+
106+
msgSeqNum2 := s.NewOrderSingle()
107+
s.fixMsgIn(s.session, msgSeqNum2)
108+
109+
s.MockApp.AssertExpectations(s.T())
110+
s.State(resendState{})
111+
s.LastToAdminMessageSent()
112+
s.MessageType(enum.MsgType_RESEND_REQUEST, s.MockApp.lastToAdmin)
113+
s.FieldEquals(tagBeginSeqNo, 1, s.MockApp.lastToAdmin.Body)
114+
s.NextTargetMsgSeqNum(1)
115+
116+
s.MessageFactory.SetNextSeqNum(1)
117+
s.MockApp.On("FromAdmin").Return(nil)
118+
s.fixMsgIn(s.session, s.SequenceReset(2))
119+
s.NextTargetMsgSeqNum(2)
120+
s.State(resendState{})
121+
122+
s.MockApp.On("FromApp").Return(nil)
123+
s.fixMsgIn(s.session, s.NewOrderSingle())
124+
125+
s.MockApp.AssertNumberOfCalls(s.T(), "FromApp", 2)
126+
s.NextTargetMsgSeqNum(4)
127+
s.State(inSession{})
128+
}

‎session.go‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,7 @@ func (s *session) sendLogout(reason string) error {
146146
return s.send(logout)
147147
}
148148

149-
func (s *session) resend(msg Message) error {
149+
func (s *session) resend(msg Message) bool {
150150
msg.Header.SetField(tagPossDupFlag, FIXBoolean(true))
151151

152152
var origSendingTime FIXString
@@ -156,12 +156,7 @@ func (s *session) resend(msg Message) error {
156156

157157
s.insertSendingTime(msg.Header)
158158

159-
if _, err := msg.Build(); err != nil {
160-
return err
161-
}
162-
s.sendBytes(msg.rawMessage)
163-
164-
return nil
159+
return s.application.ToApp(msg, s.sessionID) == nil
165160
}
166161

167162
//queueForSend will validate, persist, and queue the message for send

0 commit comments

Comments
 (0)