package org.apache.storm.messaging.jxio;


import org.apache.storm.grouping.Load;
import org.apache.storm.messaging.ConnectionWithStatus;
import org.apache.storm.messaging.IConnectionCallback;
import org.apache.storm.messaging.TaskMessage;
import org.apache.storm.messaging.org.accelio.jxio.jxioConnection.JxioConnection;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;

public class Client extends ConnectionWithStatus {

    private JxioConnection jxClient;
    private OutputStream output;
    private URI uri;
    private Object writeLock = new Object();
    private InputStream input;
    private ScheduledThreadPoolExecutor scheduler;
    private Map stormConf;

    private volatile boolean closing = false;

    Client(Map stormConf, ScheduledThreadPoolExecutor scheduler, String host, int port, Context context) {
    	this.stormConf = stormConf;
    	closing = false;
    	this.scheduler = scheduler;
    	
    	
        try {
            uri = new URI(String.format("rdma://%s:%s", host, port));
        } catch (URISyntaxException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        try {
            jxClient = new JxioConnection(uri);
            output = jxClient.getOutputStream();
            input = jxClient.getInputStream();

        } catch (ConnectException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public void registerRecv(IConnectionCallback cb) {
        // TODO Auto-generated method stub
        throw new UnsupportedOperationException("Client connection should not receive any messages");
    }

    @Override
    public void sendLoadMetrics(Map<Integer, Double> taskToLoad) {
        // TODO Auto-generated method stub
        throw new UnsupportedOperationException("Client connection should not receive any messages");

    }

    @Override
    public void send(int taskId, byte[] payload) {
        // TODO Auto-generated method stub
        TaskMessage msg = new TaskMessage(taskId, payload);
        List<TaskMessage> wrapper = new ArrayList<TaskMessage>(1);
        wrapper.add(msg);
        send(wrapper.iterator());
    }

    @Override
    public void send(Iterator<TaskMessage> msgs) {
        // TODO Auto-generated method stub

        byte[] temp = null;

        if (closing) {
            int numMessages = iteratorSize(msgs);
            return;
        }

        if (!hasMessages(msgs)) {
            return;
        }

        if (output == null) {
            dropMessages(msgs);
            return;
        }

        synchronized (writeLock) {
            while (msgs.hasNext()) {
                try {
                    output.write(msgs.next().serialize().array());
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }

    }

    private boolean hasMessages(Iterator<TaskMessage> msgs) {
        return msgs != null && msgs.hasNext();
    }

    private void dropMessages(Iterator<TaskMessage> msgs) {
        // We consume the iterator by traversing and thus "emptying" it.
        int msgCount = iteratorSize(msgs);
    }

    private int iteratorSize(Iterator<TaskMessage> msgs) {
        int size = 0;
        if (msgs != null) {
            while (msgs.hasNext()) {
                size++;
                msgs.next();
            }
        }
        return size;
    }

    @Override
    public Map<Integer, Load> getLoad(Collection<Integer> tasks) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public void close() {
        // TODO Auto-generated method stub
        jxClient.disconnect();
        closing = true;
    }

    @Override
    public Status status() {
        // TODO Auto-generated method stub
        if (closing) {
            return Status.Closed;
        }
        return Status.Connecting;
    }

    private class Connect implements Runnable{
    	
    	public Connect(){
    		
    	}
    	
    	private void reschedule(){
    		
    	}

		@Override
		public void run() {
			// TODO Auto-generated method stub
			
		}
    	
    	
    }
}

@@ -1329,32 +1329,33 @@
 ce();%0A        %7D%0A
+%0A
         try %7B%0A  
@@ -1352,28 +1352,19 @@
   try %7B%0A
-            
+%09%09%09
 jxClient
@@ -1395,113 +1395,10 @@
 i);%0A
-            output = jxClient.getOutputStream();%0A            input = jxClient.getInputStream();%0A%0A        
+%09%09
 %7D ca
@@ -1420,36 +1420,27 @@
 eption e) %7B%0A
-            
+%09%09%09
 // TODO Auto
@@ -1458,36 +1458,27 @@
 catch block%0A
-            
+%09%09%09
 e.printStack
@@ -1486,24 +1486,28 @@
 race();%0A
+%09%09%7D%0A
         
 %7D%0A    %7D%0A
@@ -1494,25 +1494,25 @@
 %09%09%7D%0A        
-%7D
+%0A
 %0A    %7D%0A%0A    
@@ -4046,32 +4046,112 @@
 necting;%0A    %7D%0A%0A
+    private boolean reconnectingAllowed() %7B%0A        return !closing;%0A    %7D%0A    %0A
     private clas
@@ -4352,16 +4352,299 @@
 stub%0A%09%09%09
+if(reconnectingAllowed())%7B%0A%09%09%09%09try %7B%0A%09%09%09%09%09output = jxClient.getOutputStream();%0A%09%09%09%09%09input = jxClient.getInputStream();%0A%09%09%09%09%7D catch (ConnectException e) %7B%0A%09%09%09%09%09// TODO Auto-generated catch block%0A%09%09%09%09%09e.printStackTrace();%0A%09%09%09%09%7D%0A%09            %0A%09%09%09%7D%0A%09%09%09else%7B%0A%09%09%09%09close();%0A%09%09%09%09return;%0A%09%09%09%7D
 %0A%09%09%7D%0A   

edit scheduler