package com.aerospike.client.command;

import com.aerospike.client.AerospikeException;
import com.aerospike.client.ScanCallback;
import com.aerospike.client.cluster.Node;
import com.aerospike.client.policy.ScanPolicy;

public final class ScanExecutor {
	
	private final ScanPolicy policy;
	private final String namespace;
	private final String setName;
	private final ScanCallback callback;
	private ScanThread[] threads;
	private Exception exception;
	
	public ScanExecutor(ScanPolicy policy, String namespace, String setName, ScanCallback callback) {
		this.policy = policy;
		this.namespace = namespace;
		this.setName = setName;
		this.callback = callback;
	}
	
	public void scanParallel(Node[] nodes)
		throws AerospikeException {
		
		threads = new ScanThread[nodes.length];
		int count = 0;
		
		for (Node node : nodes) {
			ScanThread thread = new ScanThread(this, node);
			threads[count++] = thread;
			thread.start();
		}

		for (ScanThread thread : threads) {
			try {
				thread.join();
			}
			catch (Exception e) {
			}
		}

		// Throw an exception if an error occurred.
		if (exception != null) {
			if (exception instanceof AerospikeException) {
				throw (AerospikeException)exception;		
			}
			else {
				throw new AerospikeException(exception);
			}		
		}		
	}

    private void interrupt(Exception cause)
    {
    	synchronized (this) {
    	   	if (exception != null) {
    	   		return;
    	   	}
	    	exception = cause;  		
    	}
    	
		for (ScanThread thread : threads) {
			try {
				thread.close();
				thread.interrupt();
			}
			catch (Exception e) {
			}
		}
    }

    private static final class ScanThread extends Thread {
		private final ScanExecutor parent;
		private final ScanCommand command;

		public ScanThread(ScanExecutor parent, Node node) {
			this.parent = parent;
			this.command = new ScanCommand(node, parent.callback);
		}
		
		public void run() {
			try {
				command.scan(parent.policy, parent.namespace, parent.setName);
			}
			catch (Exception e) {
				// Terminate other scan threads.
				parent.interrupt(e);
			}
		}
		
	    public void close()
	    {
	    	command.stop();
	    }		
	}
}

@@ -864,14 +864,8 @@
 ead(
-this, 
 node
@@ -1287,25 +1287,27 @@
 te void 
-interrupt
+stopThreads
 (Excepti
@@ -1512,21 +1512,26 @@
 %09thread.
-close
+stopThread
 ();%0A%09%09%09%09
@@ -1612,15 +1612,8 @@
 ate 
-static 
 fina
@@ -1664,35 +1664,25 @@
 e final 
-ScanExecutor parent
+Node node
 ;%0A%09%09priv
@@ -1681,30 +1681,24 @@
 ;%0A%09%09private 
-final 
 ScanCommand 
@@ -1731,75 +1731,223 @@
 ead(
-ScanExecutor parent, Node node) %7B%0A%09%09%09this.parent = parent;%0A%09%09%09this.
+Node node) %7B%0A%09%09%09this.node = node;%0A%09%09%7D%0A%09%09%0A%09%09public void run() %7B%0A%09%09%09try %7B%0A%09%09%09%09// Must create command here (and not in constructor), because %0A%09%09%09%09// thread local data must be initialized in the new running thread!%0A%09%09%09%09
 comm
@@ -1974,23 +1974,16 @@
 d(node, 
-parent.
 callback
@@ -1991,83 +1991,31 @@
 ;%0A%09%09
-%7D%0A%09%09%0A%09%09public void run() %7B%0A%09%09%09try %7B%0A%09%09%09%09command.scan(parent.policy, parent.
+%09%09command.scan(policy, 
 name
@@ -2021,23 +2021,16 @@
 espace, 
-parent.
 setName)
@@ -2106,24 +2106,19 @@
 %09%09%09%09
-parent.interrupt
+stopThreads
 (e);
@@ -2151,13 +2151,18 @@
 oid 
-close
+stopThread
 ()%0A%09

Create ScanCommand in same thread as it's used.