package com.llfix.api;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelUpstreamHandler;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import com.google.common.collect.ObjectArrays;
import com.llfix.FIXInitiatorPipelineFactory;
import com.llfix.IMessageCallback;
import com.llfix.IQueueFactory;
import com.llfix.util.FieldAndRequirement;
import com.llfix.util.MemoryQueueFactory;
final public class FIXInitiator {
private final String version;
private final String senderCompID;
private final String targetCompID;
private final String remoteAddress;
private final int remotePort;
private final int heartBeat;
private final boolean isDebugOn;
private final List<FieldAndRequirement> headerFields;
private final List<FieldAndRequirement> trailerFields;
private final Map<String,Channel> sessions;
private final IQueueFactory<String> queueFactory;
private final List<IMessageCallback> listeners = new ArrayList<IMessageCallback>();
private final IMessageCallback outgoingCallback;
private final ChannelHandler[] channelHandlers;
private Channel channel;
private FIXInitiator(String version, String senderCompID, String targetCompID,
String remoteAddress, int remotePort, int heartBeat, boolean isDebugOn,
List<FieldAndRequirement> headerFields,
List<FieldAndRequirement> trailerFields,
Map<String,Channel> sessions,
IQueueFactory<String> queueFactory,
IMessageCallback outgoingCallback,
ChannelHandler[] channelHandlers) {
super();
this.version = version;
this.senderCompID = senderCompID;
this.targetCompID = targetCompID;
this.remoteAddress = remoteAddress;
this.remotePort = remotePort;
this.heartBeat = heartBeat;
this.isDebugOn = isDebugOn;
this.headerFields = headerFields;
this.trailerFields = trailerFields;
this.sessions = sessions;
this.queueFactory = queueFactory;
this.outgoingCallback = outgoingCallback;
this.channelHandlers = channelHandlers;
}
public void logOn(final Map<String,String> logon){
final ClientSocketChannelFactory cf = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
final ClientBootstrap client = new ClientBootstrap(cf);
client.setPipelineFactory(new FIXInitiatorPipelineFactory(
headerFields,
trailerFields,
sessions,
queueFactory,isDebugOn,
senderCompID,
targetCompID,
outgoingCallback,
ObjectArrays.concat(channelHandlers, new MessageBroadcaster(listeners))));
final ChannelFuture channelFut = client.connect(new InetSocketAddress(remoteAddress, remotePort));
channelFut.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
System.out.println("Connection status:"+future);
channel = future.getChannel();
ChannelFuture x = channel.write(logon);
x.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
System.out.println("Write future:"+future);
}
});
}
});
}
public void logOn(){
final Map<String,String> logon = new LinkedHashMap<String, String>();
logon.put("8", version);
logon.put("56", targetCompID);
logon.put("49", senderCompID);
logon.put("35", "A");
logon.put("98", "0");
logon.put("108", Integer.toString(heartBeat));
logOn(logon);
}
public void logOff(){
final Map<String,String> logoff = new LinkedHashMap<String, String>();
logoff.put("8", version);
logoff.put("56", targetCompID);
logoff.put("49", senderCompID);
logoff.put("35", "5");
logOff(logoff);
}
public void logOff(Map<String,String> msg){
channel.write(msg);
}
public void onMsg(IMessageCallback callback){
listeners.add(callback);
}
public void sendMsg(Map<String,String> msg){
channel.write(msg);
}
public String getVersion() {
return version;
}
public String getSenderCompID() {
return senderCompID;
}
public String getTargetCompID() {
return targetCompID;
}
public String getRemoteAddress() {
return remoteAddress;
}
public int getRemotePort() {
return remotePort;
}
public Iterator<String> getAllMsgsFromStorage() throws Exception{
return queueFactory.getQueue(getSenderCompID()+"-"+getTargetCompID()).iterator();
}
public List<FieldAndRequirement> getHeaderFields() {
return headerFields;
}
public List<FieldAndRequirement> getTrailerFields() {
return trailerFields;
}
public int getHeartBeat() {
return heartBeat;
}
public static Builder Builder(String version, String senderCompID,String targetCompID, String remoteAddress, int remotePort){
return new Builder(version, senderCompID, targetCompID, remoteAddress, remotePort);
}
private static final class MessageBroadcaster implements ChannelUpstreamHandler {
private List<IMessageCallback> listeners;
public MessageBroadcaster(List<IMessageCallback> listeners) {
this.listeners = listeners;
}
@SuppressWarnings("unchecked")
@Override
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
if(e instanceof MessageEvent){
for(IMessageCallback cb : listeners){
cb.onMsg((Map<String,String>) ((MessageEvent)e).getMessage());
}
}
else if(e instanceof ExceptionEvent){
for(IMessageCallback cb : listeners){
cb.onException(((ExceptionEvent)e).getCause());
}
}
}
}
public static class Builder{
private final String version;
private final String senderCompID;
private final String targetCompID;
private final String remoteAddress;
private final int remotePort;
private int heartBeat = 30;
private boolean isDebugOn = false;
private List<FieldAndRequirement> headerFields = new ArrayList<FieldAndRequirement>();
private List<FieldAndRequirement> trailerFields = new ArrayList<FieldAndRequirement>();
private Map<String,Channel> sessions = new ConcurrentHashMap<String, Channel>();
private IQueueFactory<String> queueFactory = new MemoryQueueFactory<String>();
private ChannelHandler[] channelHandlers = new ChannelHandler[0];
private IMessageCallback outgoingCallback;
public Builder(String version, String senderCompID,String targetCompID, String remoteAddress, int remotePort) {
super();
this.version = version;
this.senderCompID = senderCompID;
this.targetCompID = targetCompID;
this.remoteAddress = remoteAddress;
this.remotePort = remotePort;
}
public Builder withAdditionalHandlers(ChannelHandler ... handlers){
channelHandlers = handlers;
return this;
}
public Builder withSessionStoreFactory(Map<String,Channel> sessions){
this.sessions = sessions;
return this;
}
public Builder withMsgStoreFactory(IQueueFactory<String> queueFactory){
this.queueFactory = queueFactory;
return this;
}
public Builder withDebugStatus(boolean isOn){
this.isDebugOn = isOn;
return this;
}
public Builder withHeartBeatSeconds(int heartBeat){
this.heartBeat = heartBeat;
return this;
}
public Builder withOutgoingMsgCallback(IMessageCallback outgoingCallback){
this.outgoingCallback = outgoingCallback;
return this;
}
public Builder withFieldRequirements(List<FieldAndRequirement> headerFields, List<FieldAndRequirement> trailerFields){
this.headerFields = headerFields;
this.trailerFields = trailerFields;
return this;
}
public FIXInitiator build(){
return new FIXInitiator(
version,
senderCompID,
targetCompID,
remoteAddress,
remotePort,
heartBeat,
isDebugOn,
headerFields,
trailerFields,
sessions,
queueFactory,
outgoingCallback,
channelHandlers);
}
}
}