import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Vector;
import java.util.List;
import java.util.concurrent.*;

import Six.Sixp.Client.*;
import Six.Sixp.Client.Error;

public class StreamingClientDemo implements Six.Sixp.Client.ISessionEventListener {
	
    private Client client;
    private ISession session;
    private MyLogger ml;
    private Vector<SixpServerSettings> servers;
	private ProxySettings proxy;
    private ReconnectStrategy reconnectStrategy;
    private int reconnectIntervalInSec;
    private List<String> prls;
    private String userId;
    private String password;
	private volatile boolean quit;
	private ConcurrentLinkedQueue<ICommand> queue;
    
    public StreamingClientDemo(Vector<SixpServerSettings> servers,
			ProxySettings proxy, ReconnectStrategy reconnectStrategy,
			int reconnectIntervalInSec, List<String> prls, String workDir,
			String userId, String password, ConcurrentLinkedQueue<ICommand> q) {
    	
    	ml = new MyLogger(workDir);
		this.servers = servers;
		this.proxy = proxy;
		this.reconnectStrategy = reconnectStrategy;
		this.reconnectIntervalInSec = reconnectIntervalInSec;
		this.userId = userId;
		this.password = password;
		this.prls = prls;
		quit = false;
		queue = q;
	}
    
    public void RunAsync()
    {
        client= new Client();
        client.SetLogging(ml);

        try {
			session = client.CreateSession(this.servers, reconnectStrategy, reconnectIntervalInSec, proxy, this);
		} catch (SixpExceptionInvalidParameter e) {
			e.printStackTrace();
			System.out.println("Invalid parameter to Create Session!");
		} catch (SixpExceptionGeneral e) {
			e.printStackTrace();
		}
        try {
			session.ConnectAsync();
		} catch (SixpExceptionConnect e) {
			e.printStackTrace();
			System.out.println("Exception on connect to the SIXP server!");
		}

        while (quit == false)
        {
			try
			{
				Thread.sleep(2000);
			}
			catch (InterruptedException e)
			{}
        }

        session.Dispose();
        client.Dispose();
        ml.Dispose();
    }
	
	public void Stop()
	{
		quit = true;
	}

	@Override
	public void OnConnected(ISession session, SixpServerSettings server, boolean reconnected) {
        ml.OnLogMessage(LogLevel.Info, "StreamingClientDemo.OnConnected");

		if (!reconnected)
		{
			ClassicLogin cl = new ClassicLogin(userId, password);
			try {
				session.LoginAsync(cl);
			} catch (SixpExceptionGeneral e) {
				System.out.println("Failed to login!");
				e.printStackTrace();
			}
		}
	}

	@Override
	public void OnConnecting(ISession session, SixpServerSettings server,
			boolean reconnecting) {
        ml.OnLogMessage(LogLevel.Info, "StreamingClientDemo.OnConnecting");
	}

	@Override
	public void OnDisconnected(ISession session, boolean unexpected) {
        System.out.println("OnDisconnected Unexpected=" + unexpected);
	}

	@Override
	public void OnError(ISession session, SixpServerSettings server, Error errorCode,
			String errorMessage) {
        ml.OnLogMessage(LogLevel.Info, "StreamingClientDemo.OnError ec:" + errorCode + " em:" + errorMessage);
        System.out.println("OnError ec:" + errorCode + " em:" + errorMessage );		
	}

	@Override
	public void OnLoggedIn(ISession session, boolean reconnected) {
        ml.OnLogMessage(LogLevel.Info, "StreamingClientDemo.OnLoggedIn");
        if (!reconnected)
        {
			DataListener dl = new DataListener(session, queue);
            for (String prl : prls)
            {
				queue.add(new TableAdder(prl));
				IRequest request = InitializeDataListerner(session, dl, prl);
                
				if (request != null)
				{
					request.ExecuteAsync();
				}
            }
        }
	}

	@Override
	public void OnLoggedOut(ISession session, SixpServerSettings server, Error errorCode,
			String errorMessage) {
        ml.OnLogMessage(LogLevel.Info, "StreamingClientDemo.OnLoggeOut ec:" + errorCode + " em:" + errorMessage);
	}

	@Override
	public void OnTimingInfo(ISession session, long pingTurnaroundTime, long estimatedDelayTime, long queuesize) {
		//System.out.println("ping: "+ pingTurnaroundTime + " delay: "+ estimatedDelayTime + "queuesize: " + queuesize);
	}
	
	private IRequest InitializeDataListerner(ISession session, DataListener dl, String prl)
	{
		IRequest request = null;
        try {
			System.out.println("Setup prl: " + prl);
			request = session.CreateRequest(prl, AcceptFormat.Table, true, dl);
		} catch (SixpExceptionNotLoggedIn e) {
			e.printStackTrace();
			System.out.println("Not logged into SIXP!");
		} catch (SixpExceptionInvalidParameter e) {
			e.printStackTrace();
			System.out.println("Invalid parameter!");
		}
		return request;
	}
}
