import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import models.IronArgusFlow;
import models.IronMitigationEvent;

import org.apache.http.client.ClientProtocolException;
import org.atmosphere.annotation.Broadcast;
import org.atmosphere.cpr.Broadcaster;
import org.atmosphere.jersey.Broadcastable;
import org.atmosphere.jersey.SuspendResponse;

import com.ironnet.Hermes;
import com.ironnet.exceptions.PublishException;
import com.ironnet.models.type.IronType;

import Runnables.ArgusFlowRunnable;
import Runnables.IronMitigationRunnable;
import Runnables.IronScoreBaselineRunnable;
import Runnables.IronScoreClusterRunnable;

import javax.ws.rs.FormParam;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;

/**
 * Simple PubSub resource that demonstrate many functionality supported by
 * Atmosphere Javascript and Atmosphere Jersey extension.
 *
 * @author Jeanfrancois Arcand
 */
@Path("/pubsub/{topic}")
public class JerseyPubSub {

    private
    @PathParam("topic")
    Broadcaster topic;

    @GET
    public SuspendResponse<String> subscribe() {
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import models.IronArgusFlow;
import models.IronMitigationEvent;

import org.apache.http.client.ClientProtocolException;
import org.atmosphere.annotation.Broadcast;
import org.atmosphere.cpr.Broadcaster;
import org.atmosphere.jersey.Broadcastable;
import org.atmosphere.jersey.SuspendResponse;

import com.ironnet.Hermes;
import com.ironnet.exceptions.PublishException;
import com.ironnet.models.type.IronType;

import Runnables.ArgusFlowRunnable;
import Runnables.IronMitigationRunnable;
import Runnables.IronScoreBaselineRunnable;
import Runnables.IronScoreClusterRunnable;

import javax.ws.rs.FormParam;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;

/**
 * Simple PubSub resource that demonstrate many functionality supported by
 * Atmosphere Javascript and Atmosphere Jersey extension.
 *
 * @author Jeanfrancois Arcand
 */
@Path("/pubsub/{topic}")
public class JerseyPubSub {

    private
    @PathParam("topic")
    Broadcaster topic;

    @GET
    public SuspendResponse<String> subscribe() {

    	Broadcaster b = topic;

	//    	ArgusFlowRunnable afr = new ArgusFlowRunnable(topic);
     	IronScoreClusterRunnable iscr = new IronScoreClusterRunnable(topic);
	//	IronScoreBaselineRunnable isbr = new IronScoreBaselineRunnable(topic);
	//	IronMitigationRunnable imr = new IronMitigationRunnable (topic);

	//	b.getBroadcasterConfig().getExecutorService().execute(afr);
    	b.getBroadcasterConfig().getExecutorService().execute(iscr);
	//  	b.getBroadcasterConfig().getExecutorService().execute(isbr);
	//	b.getBroadcasterConfig().getExecutorService().execute(imr);


        return new SuspendResponse.SuspendResponseBuilder<String>()
                .broadcaster(topic)
                .outputComments(true)
                .addListener(new EventsLogger())
                .build();
    }

    @POST
    @Broadcast
    @Produces("text/json")
    public Broadcastable publish(String message) {

    	 IronType type;
		try {
			type = Hermes.createIronType(IronMitigationEvent.getSchema());
		} catch (ClientProtocolException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

         // Set up to get the topic in the KV
         if (Hermes.getIronTopic("mitigation_recommended") == null)
         {
             try {
				Hermes.createIronTopic("mitigation_recommended", type.getGuid(), null);
			} catch (ClientProtocolException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
         }

         long start = System.currentTimeMillis();

         // The server should've already defined the "triplet list" for the
         // data they will be sending.  We are sending ServerUser objects, and
         // have defined the triplets within that class.
         for (int i = 0; i < 1000; i++)
         {
             IronArgusFlow flow = new IronArgusFlow();
             flow.timeSinceEpoch = System.currentTimeMillis();
             flow.saddr = Integer.toString(i);

              System.out.println(flow.saddr);
             // A static method on the data type seems like an easy way
             // to always have access to the triplet list.
             //
             // "testtopic" just happens to be the topic we are producing on,
             // but in the real scenario, we would've created a new topic
             // with a GUID and called "createTopic".  This is just here to
             // show how to publish data sans KV
             try
             {
                 Hermes.publishData(type, flow, "argus_flow");
             }
             catch (IOException | PublishException e)
             {
                 e.printStackTrace();
             }
         }

         long end = System.currentTimeMillis();

         System.out.println("Time elapsed: " + (end - start));

    	System.out.println("recvd: " + message);
        return new Broadcastable(message, "", topic);
    }
}

    	Broadcaster b = topic;

	//    	ArgusFlowRunnable afr = new ArgusFlowRunnable(topic);
     	IronScoreClusterRunnable iscr = new IronScoreClusterRunnable(topic);
	//	IronScoreBaselineRunnable isbr = new IronScoreBaselineRunnable(topic);
	//	IronMitigationRunnable imr = new IronMitigationRunnable (topic);

	//	b.getBroadcasterConfig().getExecutorService().execute(afr);
    	b.getBroadcasterConfig().getExecutorService().execute(iscr);
	//  	b.getBroadcasterConfig().getExecutorService().execute(isbr);
	//	b.getBroadcasterConfig().getExecutorService().execute(imr);


        return new SuspendResponse.SuspendResponseBuilder<String>()
                .broadcaster(topic)
                .outputComments(true)
                .addListener(new EventsLogger())
                .build();
    }

    @POST
    @Broadcast
    @Produces("text/json")
    public Broadcastable publish(String message) {

    	 IronType type;
		try {
			type = Hermes.createIronType(IronMitigationEvent.getSchema());
		} catch (ClientProtocolException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

         // Set up to get the topic in the KV
         if (Hermes.getIronTopic("mitigation_recommended") == null)
         {
             try {
				Hermes.createIronTopic("mitigation_recommended", type.getGuid(), null);
			} catch (ClientProtocolException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
         }

         long start = System.currentTimeMillis();

         // The server should've already defined the "triplet list" for the
         // data they will be sending.  We are sending ServerUser objects, and
         // have defined the triplets within that class.
         for (int i = 0; i < 1000; i++)
         {
             IronArgusFlow flow = new IronArgusFlow();
             flow.timeSinceEpoch = System.currentTimeMillis();
             flow.saddr = Integer.toString(i);

              System.out.println(flow.saddr);
             // A static method on the data type seems like an easy way
             // to always have access to the triplet list.
             //
             // "testtopic" just happens to be the topic we are producing on,
             // but in the real scenario, we would've created a new topic
             // with a GUID and called "createTopic".  This is just here to
             // show how to publish data sans KV
             try
             {
                 Hermes.publishData(type, flow, "argus_flow");
             }
             catch (IOException | PublishException e)
             {
                 e.printStackTrace();
             }
         }

         long end = System.currentTimeMillis();

         System.out.println("Time elapsed: " + (end - start));

    	System.out.println("recvd: " + message);
        return new Broadcastable(message, "", topic);
    }
}