package de.fernunihagen.dna.jkn.scalephant.distribution.sstable;

import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.jkn.scalephant.ScalephantConfiguration;
import de.fernunihagen.dna.jkn.scalephant.ScalephantConfigurationManager;
import de.fernunihagen.dna.jkn.scalephant.distribution.DistributionRegion;
import de.fernunihagen.dna.jkn.scalephant.distribution.membership.DistributedInstance;
import de.fernunihagen.dna.jkn.scalephant.distribution.membership.DistributedInstanceManager;
import de.fernunihagen.dna.jkn.scalephant.distribution.resource.RandomResourcePlacementStrategy;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.FloatInterval;

public class SimpleSplitStrategy extends SSTableSplitter {
	
	/**
	 * The Logger
	 */
	protected final static Logger logger = LoggerFactory.getLogger(SimpleSplitStrategy.class);

	/**
	 * Test if a split is needed
	 */
	@Override
	public boolean isSplitNeeded(final int totalTuplesInTable) {
		final ScalephantConfiguration configuration = ScalephantConfigurationManager.getConfiguration();
		final int maxEntries = configuration.getSstableMaxEntries();
		
		if(totalTuplesInTable > maxEntries) {
			return true;
		} else {
			return false;
		}
	}

	/**
	 * Perform a split of the given sstable
	 */
	@Override
	protected void performSplit(final DistributionRegion region) {
		
		final DistributedInstanceManager distributedInstanceManager = DistributedInstanceManager.getInstance();
		final Set<DistributedInstance> systems = distributedInstanceManager.getInstances();
		
		if(systems.isEmpty()) {
			logger.warn("Unable to split region, no ressources are avilable: " + region);
			return;
		}
		
		logger.info("Performing split of region: " + region);
		
		// Split region
		final int splitDimension = region.getSplitDimension();
		final FloatInterval interval = region.getConveringBox().getIntervalForDimension(splitDimension);
		
		logger.info("Split at dimension:" + splitDimension + " interval: " + interval);
		float midpoint = interval.getMidpoint();
		
		logger.info("Set split at:" + midpoint);
		region.setSplit(midpoint);
		
		// Assign systems
		final RandomResourcePlacementStrategy resourcePlacementStrategy = new RandomResourcePlacementStrategy();
		
		final DistributedInstance systemLeftChild = resourcePlacementStrategy.findSystemToAllocate(systems);
		final DistributedInstance systemRightChild = resourcePlacementStrategy.findSystemToAllocate(systems);
		
		region.getLeftChild().addSystem(systemLeftChild);
		region.getRightChild().addSystem(systemRightChild);
	}
}

@@ -352,24 +352,250 @@
 tionRegion;%0A
+import de.fernunihagen.dna.jkn.scalephant.distribution.ZookeeperClient;%0Aimport de.fernunihagen.dna.jkn.scalephant.distribution.ZookeeperClientFactory;%0Aimport de.fernunihagen.dna.jkn.scalephant.distribution.ZookeeperException;%0A
 import de.fe
@@ -2716,60 +2716,241 @@
 %09%0A%09%09
-region.getLeftChild().addSystem(systemLeftChild);%0A%09%09
+try %7B%0A%09%09%09final ZookeeperClient zookeeperClient = ZookeeperClientFactory.getZookeeperClient();%0A%09%09%09zookeeperClient.addSystemToDistributionRegion(region.getLeftChild(), systemLeftChild);%0A%09%09%09zookeeperClient.addSystemToDistributionRegion(
 regi
@@ -2971,19 +2971,10 @@
 ld()
-.addSystem(
+, 
 syst
@@ -2988,13 +2988,132 @@
 Child);%0A
+%09%09%7D catch (ZookeeperException e) %7B%0A%09%09%09logger.warn(%22Unable to assign systems to splitted region: %22 + region, e);%0A%09%09%7D%0A%09%09%0A
 %09%7D%0A%7D%0A

Write splits to zookeeper