changed IoSessionResponder to use session log

Author: chrjohn

quickfixj-core/src/main/java/quickfix/mina/IoSessionResponder.java

@@ -21,8 +21,7 @@
 
 import org.apache.mina.core.future.WriteFuture;
 import org.apache.mina.core.session.IoSession;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import quickfix.LogUtil;
 import quickfix.Responder;
 import quickfix.Session;
 
@@ -34,7 +33,6 @@
  * the MINA networking code.
  */
 public class IoSessionResponder implements Responder {
-    private final Logger log = LoggerFactory.getLogger(getClass());
     private final IoSession ioSession;
     private final boolean synchronousWrites;
     private final long synchronousWriteTimeout;
@@ -50,8 +48,8 @@ public IoSessionResponder(IoSession session, boolean synchronousWrites, long syn
     @Override
     public boolean send(String data) {
         // Check for and disconnect slow consumers.
+        Session qfjSession = (Session) ioSession.getAttribute(SessionConnector.QF_SESSION);
         if (maxScheduledWriteRequests > 0 && ioSession.getScheduledWriteMessages() >= maxScheduledWriteRequests) {
-            Session qfjSession = (Session) ioSession.getAttribute(SessionConnector.QF_SESSION);
             try {
                 qfjSession.disconnect("Slow consumer", true);
             } catch (IOException e) {
@@ -64,11 +62,11 @@ public boolean send(String data) {
         if (synchronousWrites) {
             try {
                 if (!future.awaitUninterruptibly(synchronousWriteTimeout)) {
-                    log.error("Synchronous write timed out after {}ms", synchronousWriteTimeout);
+                    qfjSession.getLog().onErrorEvent("Synchronous write timed out after " + synchronousWriteTimeout + "ms.");
                     return false;
                 }
             } catch (RuntimeException e) {
-                log.error("Synchronous write failed: {}", e.getMessage());
+                LogUtil.logThrowable(qfjSession.getSessionID(), "Synchronous write failed.", e);
                 return false;
             }
         }

quickfixj-core/src/test/java/quickfix/mina/IoSessionResponderTest.java

@@ -34,9 +34,14 @@
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
+import quickfix.Log;
+import quickfix.Session;
+import quickfix.SessionID;
 
 public class IoSessionResponderTest {
 
+    private final SessionID sessionID = new SessionID("FIX.4.4:SENDER->TARGET");
+
     @Test
     public void testAsynchronousSend() throws Exception {
         IoSession mockIoSession = mock(IoSession.class);
@@ -48,6 +53,7 @@ public void testAsynchronousSend() throws Exception {
         boolean result = responder.send("abcd");
 
         assertTrue(result);
+        verify(mockIoSession).getAttribute(SessionConnector.QF_SESSION);
         verify(mockIoSession).write("abcd");
         verifyNoMoreInteractions(mockWriteFuture);
         verifyNoMoreInteractions(mockIoSession);
@@ -65,6 +71,7 @@ public void testSynchronousSend() throws Exception {
         boolean result = responder.send("abcd");
 
         assertTrue(result);
+        verify(mockIoSession).getAttribute(SessionConnector.QF_SESSION);
         verify(mockIoSession).write("abcd");
         verify(mockWriteFuture).awaitUninterruptibly(timeout);
         verifyNoMoreInteractions(mockWriteFuture);
@@ -75,7 +82,10 @@ public void testSynchronousSend() throws Exception {
     public void testSynchronousSendWithJoinException() throws Exception {
         int timeout = 123;
         IoSession mockIoSession = mock(IoSession.class);
-
+        Session mockSession = mock(Session.class);
+        when(mockIoSession.getAttribute(SessionConnector.QF_SESSION)).thenReturn(mockSession);
+        when(mockSession.getSessionID()).thenReturn(sessionID);
+        
         WriteFuture mockWriteFuture = mock(WriteFuture.class);
         when(mockIoSession.write("abcd")).thenReturn(mockWriteFuture);
         doThrow(new RuntimeException("TEST")).when(mockWriteFuture).awaitUninterruptibly(timeout);
@@ -84,6 +94,7 @@ public void testSynchronousSendWithJoinException() throws Exception {
         boolean result = responder.send("abcd");
 
         assertFalse(result);
+        verify(mockIoSession).getAttribute(SessionConnector.QF_SESSION);
         verify(mockIoSession).write("abcd");
         verify(mockWriteFuture).awaitUninterruptibly(timeout);
         verifyNoMoreInteractions(mockWriteFuture);
@@ -94,6 +105,11 @@ public void testSynchronousSendWithJoinException() throws Exception {
     public void testSynchronousSendWithJoinTimeout() throws Exception {
         int timeout = 123;
         IoSession mockIoSession = mock(IoSession.class);
+        Session mockSession = mock(Session.class);
+        Log log = mock(Log.class);
+        when(mockIoSession.getAttribute(SessionConnector.QF_SESSION)).thenReturn(mockSession);
+        when(mockSession.getSessionID()).thenReturn(sessionID);
+        when(mockSession.getLog()).thenReturn(log);
 
         WriteFuture mockWriteFuture = mock(WriteFuture.class);
         when(mockIoSession.write("abcd")).thenReturn(mockWriteFuture);
@@ -103,6 +119,7 @@ public void testSynchronousSendWithJoinTimeout() throws Exception {
         boolean result = responder.send("abcd");
 
         assertFalse(result);
+        verify(mockIoSession).getAttribute(SessionConnector.QF_SESSION);
         verify(mockIoSession).write("abcd");
         verify(mockWriteFuture).awaitUninterruptibly(timeout);
         verifyNoMoreInteractions(mockWriteFuture);