import core.thread;
import std.socket;
import std.stream;
import std.socketstream;
import std.stdio; 

/**
   An experimental TCP/IP socket server, written in the C++/Java subset of D.

   Author: Ian Kaplan
 */

/**
   The stream server has a simple protocol with the client. It expects the client to tell it
   the name of the file containing the time series. The server will then open the file and
   stream it to the client.

   The command is: "open:<fileName>"  For example: "open:GS.Jul_01_2008", where GS.Jul_01_2008 is the
   name of the file.  If the file is opened successfully, the server will respond with "OK\n". If there
   is a problem opening the file the server will respond with "ERROR:<message>".  The stream will be
   closed if there is an error. 
  
   Following the OK will be a stream of values and time stamps.
 */
class StreamServer : Thread 
{
	private string dataPath;
	private Socket socket;
    private Stream ioStream;
    private uint lastMsec = 0;
    private bool processStream = true;
	
	this( string path, Socket soc )
	{
		super( &run );
		dataPath = path;
		socket = soc;
		ioStream = new SocketStream( soc );
	} // constructor
	
	void stopProcessing()
	{
		processStream = false;
	}
	
	/**
	   Strip the file name out of the "open:<filename>" string.  There must be a better 
	   way, but I could not figure it out.  This is a pretty crude algorith. It just looks
	   for the ':' and then copies the characters after that.
	 */
	static string getFileName( char[] line )
	{
		string fileName;
		bool start = false;
		foreach (ch; line) {
			if (start) {
				fileName ~= ch;
			}
			if (ch == ':') {
				start = true;
			}
		}
		return fileName;
	} // getFileName
	
	/**
	   Convert a string of decimal digit characters into an unsigned integer. This
	   code does no format checking.  There is certainly a better way to do this in 
	   D, but I was in a rush to get this code done.
	 */
	private uint toInt( char[] numChars )
	{
		uint val = 0;
		foreach (ch; numChars) {
			val = (val * 10) + (ch - '0');
		}
		return val;
	} // toInt
	
	/**
	    The time series values that are streamed by the StreamServer have the
	    format <floating point value>,<time>
	    The time field has the format:  <hr>:<min>:<sec>:<msec>
	    This code is written with the assumption that the hour between two sequential
	    time series values is constant (obviously not a general, robust assumption).
	    Also, there's no format checking. This code assumes that the values are properly
	    formed (again, not the best assumption).
	
	    I'm still not proficient at D strings, so this may not be the best of algorithms.
	 */
	private uint getWaitTime( char[] line )
	{
		enum State { TSVal, Hour, Min, Sec, MSec };
		State curState = State.TSVal;
		char[] min;
		char[] sec;
		char[] msec;
		uint waitMSec = 0;
		foreach (ch; line) {
		   switch (curState) {
		   		case State.TSVal:
		   		if (ch == ',') {
		   			curState = State.Hour;
		   		}
		   		break;
		   		case State.Hour:
		   		if (ch == ':') {
		   			curState = State.Min;
		   		}
		   		break;
		   		case State.Min:   // Minutes
		   		if (ch == ':') {
		   			curState = State.Sec;
		   		}
		   		else {
		   			min ~= ch;
		   		}
		   		break;
		   		case State.Sec:   // Seconds
		   		if (ch == ':') {
		   			curState = State.MSec;
		   		} 
		   		else {
		   			sec ~= ch;
		   		}
		   		break; 
		   		case State.MSec:  // Milliseconds
		   		msec ~= ch;
		   		break;
		   		default: // This should never happen
		   		break;
		   } // switch
		} // foreach
		uint minVal = toInt( min );
		uint secVal = toInt( sec ); 
		uint msecVal = toInt( msec );
		uint totalMSec = minVal * 60 * 1000 + secVal * 1000 + msecVal; 
		if (lastMsec == 0) {
			lastMsec = totalMSec;
		}
        else {
        	waitMSec = totalMSec - lastMsec;
        	lastMsec = totalMSec;
        }
		return waitMSec;
	}  // getWaitTime

	/**
	   Stream the contents of a time stamped time series file. This code attempts to 
	   emulate the streaming time in the time series by inserting the proper wait times.
	 */
	private void streamFile( std.stream.File file )
	{
		char[] line;
		do {
			line = file.readLine();
			if (line.length > 0) {
				uint millisec = getWaitTime( line );
				if (processStream && millisec > 0) {
					Thread.sleep( dur!("msecs")( millisec ));
				}
				if (processStream) {
					ioStream.writeLine( line );
				}
			}
		} while (processStream && line.length > 0);
	} // streamFile
	
	
	private void handleStream()
	{
		// read the open command
		char[] line = ioStream.readLine();
		if (line.length > 0) {
			bool isOK = true;
			string fileName = getFileName( line );
			string response = "OK";
			if (fileName.length == 0) {
				response = "ERROR: zero length file name";
				isOK = false;
			}
			std.stream.File f;
			if (isOK) {
				string filePath = dataPath ~ '/' ~ fileName;
				f = new std.stream.File(filePath, FileMode.In );
				if (! f.isOpen() ) {
					response = "ERROR: error opening file: " ~ filePath;
					isOK = false;
				}
			}
			ioStream.writeLine( response );
			if (isOK) {
				streamFile( f );
				ioStream.close();
				f.close();
			}
		}
		else {
			writeln("StreamServer: zero length line received. Closing the socket");
			socket.close();
		}
	}
	
	
	private void run()
	{
		handleStream();
	} // run
} // class StreamServer


/**
  This is the main thread of the socket server, which connects to the clients.
*/
class SocketServer : Thread {
    immutable serverPort = 4256;  // connection port for the socket for the server
    
    private immutable maxWaitingConnects = 10;  // maximum number of waiting connections allowed
    private immutable maxConnects = 60;         // total number of connections
    private Socket listener;
    private SocketSet sset;
    private Socket[] reads;
    private bool bindOK = false;
    private bool keepListening = true;
    private string dataPath;  // path to the directory containing the time series data.
    private StreamServer[] serverList;
    
    this(string path)
    {
    	super( &run );
    	dataPath = path;
    	listener = new TcpSocket;
    	sset = new SocketSet(maxConnects + 1);
    }
    
    void stopListening()
    {
    	keepListening = false;
    }
        
    bool socketBind()
    {
    	if (listener.isAlive()) {
    		listener.blocking( false );
    		listener.bind(new InternetAddress( serverPort ) );
    		bindOK = true;
    		
    	}
    	return bindOK;
    } // socketBind
    
    void listen()
    {
    	if (bindOK) {
    		listener.listen( maxWaitingConnects );
    		writefln("Listening on port %d", serverPort );
    		Socket sn;
    		while (keepListening) {
    			sn = listener.accept();
    			if (sn.isAlive()) {
    				writefln("Connection from %s established", sn.remoteAddress().toString() );
    				StreamServer strServ = new StreamServer(dataPath, sn );
    				serverList ~= strServ;
    				strServ.start();
    			}
    		}
    	}
    	else {
    		writeln("listen: socket was not bound successfully before calling listen()");
    	}
    	foreach (serv; serverList) {
    		serv.stopProcessing();
    	}
    	writeln("Leaving listen()");
    } // listen
    
    private void run()
    {
    	writeln("Starting listen thread");
    	listen();
    }
} // class SocketServer


/**
    Stream a single time series from a file.
 */
class TimeSeriesClient : Thread
{
	private Address mHostAddr;
	private string mFileName;
	private bool mContinue = true;
	
	this(string fileName)
	{
		super( &run );
		mFileName = fileName;
		Address[] addresses = getAddress("localhost", SocketServer.serverPort );
		mHostAddr = addresses[0];
	}
	
	void stopClient()
	{
		mContinue = false;
	}
	
	/**
	  Allocate a TCP/IP socket and connect to the host address. The function returns the
	  bound socket. The socket must be examined by the caller to determine if there was an
	  error in the connection.
	 */
	Socket connectSocket( Address hostAddr )
	{
		Socket clientSock = new TcpSocket();
		clientSock.connect( hostAddr );
	    return clientSock;	
	}
	
	/**
	   Obtain a socket connection to a time series and read the time series stream.
	*/
	private void run()
	{
	    Socket sn = connectSocket( mHostAddr );
	    SocketStream socStream = new SocketStream( sn );
	    string openStr = "open:" ~ mFileName;
	    socStream.writeLine( openStr );
	    char[] res = socStream.readLine();
	    if (res.length > 0) {
	    	if (res == "OK") {
	    		char[] line;
	    		do {
	    			if (mContinue) {
	    				line = socStream.readLine();
	    				if (line.length > 0) {
	    				   writefln("%s: %s", mFileName, line );
	    				   stdout.flush();
    				   }
    				}
	    		} while (mContinue && line.length > 0);
	    		socStream.close(); 
	    	}
	    	else {
	    		writeln( res );
	    	}
	    }
	    else {
	    	writeln("socket response has zero length");
	    }		
	}
} // class TimeSeriesClient




void main() 
{ 
	immutable path = "C:/Users/kaplan4/Documents/finance/market_data";
	writeln("Hello from the main of the socket server program");
	SocketServer serv = new SocketServer( path );
	if (serv.socketBind()) {
		serv.start();  // Start the socket server
		// The names of the time series files
        string fileNames[] = [ 
                             "AMGN.Jul_02_2008", 
                             "AMZN.Jul_02_2008",
                             "BBBY.Jul_02_2008", 
                             "BIIB.Jul_02_2008",
                             "BRCM.Jul_02_2008",
                             "CAT.Jul_02_2008",
                             "CME.Jul_02_2008",
                             "CMI.Jul_02_2008",
                             "ERTS.Jul_02_2008",
                             "FCX.Jul_02_2008",
                             "GENZ.Jul_02_2008",
                             "GOOG.Jul_02_2008",
                             "GS.Jul_02_2008", 
                             "ICE.Jul_02_2008",
                             "OIH.Jul_02_2008",
                             "RIG.Jul_02_2008"
                              ];
        // build a list of client objects                              
        TimeSeriesClient clientList[];
        foreach (name; fileNames)
        {
        	TimeSeriesClient client = new TimeSeriesClient( name );
        	clientList ~= client;
        }  
        // start the clients
        foreach (client; clientList) {
        	client.start();
        }                       
	}
}

unittest {
	writeln("Hello from the unittest");
	assert( SocketServer.serverPort == 4256);
}