This patch ends dispatcher thread after logout.

/*******************************************************************************

• Copyright (c) quickfixengine.org All rights reserved.
  *
• This file is part of the QuickFIX FIX Engine
  *
• This file may be distributed under the terms of the quickfixengine.org
• license as defined by quickfixengine.org and appearing in the file
• LICENSE included in the packaging of this file.
  *
• This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING
• THE WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A
• PARTICULAR PURPOSE.
  *
• See http://www.quickfixengine.org/LICENSE for licensing information.
  *
• Contact [email protected] if any conditions of this licensing
• are not clear to you.
  ******************************************************************************/

package quickfix.mina;

import quickfix.LogUtil;
import quickfix.Message;
import quickfix.Session;
import quickfix.SessionID;
import quickfix.SessionStateListener;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**

• Processes messages in a session-specific thread.
  */
  public class ThreadPerSessionEventHandlingStrategy implements EventHandlingStrategy {
  private final Map<SessionID, MessageDispatchingThread> dispatchers = new ConcurrentHashMap<SessionID, MessageDispatchingThread>();

public void onMessage(Session quickfixSession, Message message) {
MessageDispatchingThread dispatcher = dispatchers.get(quickfixSession.getSessionID());
if (dispatcher == null)

dispatcher.enqueue(message);
}

/** There is no such thing as a SesionConnector for thread-per-session handler - we don't multiplex

• between multiple sessions here so this is null
  */
  public SessionConnector getSessionConnector() { return null; }

protected void startDispatcherThread(MessageDispatchingThread dispatcher)

public void stopDispatcherThreads() {
Collection<MessageDispatchingThread> dispatchersToShutdown = dispatchers.values();
dispatchers.clear();
for (MessageDispatchingThread dispatcher : dispatchersToShutdown)

}

private final class MessageDispatchingThread extends Thread {
private final Session quickfixSession;
private final BlockingQueue<Message> messages = new LinkedBlockingQueue<Message>();
private volatile boolean stopped;

private MessageDispatchingThread(Session session) {
super("QF/J Session dispatcher: " + session.getSessionID());
quickfixSession = session;
quickfixSession.addStateListener(new SessionStateListener() {
public void onConnect() {
}

public void onDisconnect() {
}

public void onLogon() {
}

public void onLogout()

public void onReset() {
}

public void onRefresh() {
}

public void onMissedHeartBeat() {
}

public void onHeartBeatTimeout() {
}
});
}

public void enqueue(Message message) {
try

catch (InterruptedException e)

}

public int getQueueSize()

public void run() {
while (!stopped) {
try {
Message message = getNextMessage(messages);
if (quickfixSession.hasResponder())

} catch (InterruptedException e) {
if (!stopped)

} catch (Throwable e)

}
}

public void stopDispatcher()

}

private BlockingQueue<Message> getMessages(SessionID sessionID)

private MessageDispatchingThread getDispatcher(SessionID sessionID)

private Message getNextMessage(BlockingQueue<Message> messages) throws InterruptedException

public int getQueueSize() {
int ret = 0;
for(MessageDispatchingThread mdt : dispatchers.values())

return ret;
}

}