package net.floodlightcontroller.nfvtest.nfvslaveservice;

import java.util.concurrent.LinkedBlockingQueue;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

import net.floodlightcontroller.nfvtest.message.Message;
import net.floodlightcontroller.nfvtest.message.MessageHub;
import net.floodlightcontroller.nfvtest.message.MessageProcessor;
import net.floodlightcontroller.nfvtest.message.ConcreteMessage.KillSelfRequest;

import net.floodlightcontroller.nfvtest.message.ConcreteMessage.*;

public class SubscriberConnector extends MessageProcessor{
	private final Context zmqContext;
	
	
	public SubscriberConnector(String id, Context zmqContext){
		this.id = id;
		this.queue = new LinkedBlockingQueue<Message>();
		this.zmqContext = zmqContext;
	}
	
	@Override
	public void run() {
		// TODO Auto-generated method stub
		LinkedBlockingQueue<Message> q = (LinkedBlockingQueue<Message>)this.queue;
		while(true){
			try{
				Message m = q.take();
				if(m instanceof KillSelfRequest){
					break;
				}
				else{
					onReceive(m);
				}
			}
			catch (Exception e){
				e.printStackTrace();
			}			
		}
	}

	@Override
	protected void onReceive(Message m) {
		// TODO Auto-generated method stub
		if(m instanceof SubConnRequest){
			SubConnRequest request = (SubConnRequest)m;
			subscriberConnect(request);
		}
	}
	
	public class SubConnThread implements Runnable{
		private final SubConnRequest request;
		private final Context zmqContext;
		private final MessageHub mh;
		
		public SubConnThread(SubConnRequest request, Context zmqContext, MessageHub mh){
			this.request = request;
			this.zmqContext = zmqContext;
			this.mh = mh;
		}
		
		@Override
		public void run() {
			// TODO Auto-generated method stub
			try{
				Thread.sleep(5000);
			}
			catch (Exception e){
				e.printStackTrace();
			}
			
			String ipAddress = this.request.getManagementIp();
			String port = this.request.getPort();
			
			Socket subscriber = zmqContext.socket(ZMQ.SUB);
			subscriber.monitor("inproc://monitor"+ipAddress, ZMQ.EVENT_CONNECTED);
			
			Socket monitor = zmqContext.socket(ZMQ.PAIR);
			monitor.setReceiveTimeOut(100);
			monitor.connect("inproc://monitor"+ipAddress);
			ZMQ.Event event;
			
			//retry 10 times
			for(int i=0; i<10; i++){
				subscriber.connect("tcp://"+ipAddress+":"+port);
	        	event = ZMQ.Event.recv(monitor);
	        	
	        	if((event != null)&&(event.getEvent() == ZMQ.EVENT_CONNECTED)){
	        		subscriber.subscribe("".getBytes());
	        		
	        		SubConnReply reply = new SubConnReply("hehe", this.request, subscriber);
	        		mh.sendTo(this.request.getSourceId(), reply);
	        		
	        		break;
	        	}
	        	else{
	        	}
			}
		}
	}
	
	private void subscriberConnect(SubConnRequest request){
		SubConnThread newThread = new SubConnThread(request, this.zmqContext, this.mh);
		Thread t = new Thread(newThread);
		t.start();
	}
	
}

@@ -1935,24 +1935,55 @@
 Port();%0A%09%09%09%0A
+%09%09%09for(int i=0; i%3C1000; i++)%7B%0A%09
 %09%09%09Socket su
@@ -2021,16 +2021,17 @@
 Q.SUB);%0A
+%09
 %09%09%09subsc
@@ -2103,16 +2103,17 @@
 %0A%09%09%09%0A%09%09%09
+%09
 Socket m
@@ -2149,24 +2149,25 @@
 Q.PAIR);%0A%09%09%09
+%09
 monitor.setR
@@ -2186,16 +2186,17 @@
 t(100);%0A
+%09
 %09%09%09monit
@@ -2240,16 +2240,17 @@
 ss);%0A%09%09%09
+%09
 ZMQ.Even
@@ -2266,56 +2266,8 @@
 %09%09%09%0A
-%09%09%09//retry 10 times%0A%09%09%09for(int i=0; i%3C10; i++)%7B%0A
 %09%09%09%09
@@ -2702,16 +2702,75 @@
  %09else%7B%0A
+%09        %09%09monitor.close();%0A%09        %09%09subscriber.close();%0A
 %09       

change ServiceChainHandler so that the subscriber won't connect to the publisher multiple times