<commit_before>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{<commit_msg>change ServiceChainHandler so that the subscriber won't connect to the publisher multiple times<commit_after>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();
			
			for(int i=0; i<1000; i++){
				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;
			
				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{
	        		monitor.close();
	        		subscriber.close();
{'fraction_non_alphanumeric': 0.11008358527476436, 'fraction_numerical': 0.004268184243286502, 'mean_word_length': 3.4599524187153055, 'pattern_counts': {'":': 2, '<': 11, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2501930', 'n_tokens_mistral': 2075, 'n_tokens_neox': 1729, 'n_words': 348}