// Path: reference/src/main/java/me/doubledutch/stroom/client/StreamConnection.java
// public interface StreamConnection{
// 	public final static int LINEAR=0;
// 	public final static int SYNC=1;
// 	public final static int FLUSH=2;
// 	public final static int NONE=3;
// 
// 	public List<String> get(long index,long endIndex) throws IOException;
// 	public String get(long index) throws IOException;
// 	public String getLast() throws IOException;
// 	public long getCount() throws IOException;
// 	
// 	public long append(JSONObject data) throws IOException,JSONException;
// 	public long append(String data) throws IOException,JSONException;
// 	public List<Long> append(List<String> data) throws IOException,JSONException;
// 	
// 	public long append(JSONObject data,int hint) throws IOException,JSONException;
// 	public long append(String data,int hint) throws IOException,JSONException;
// 	public List<Long> append(List<String> data,int hint) throws IOException,JSONException;
// 
// 	public void truncate(long index) throws IOException;
// }


import org.apache.log4j.Logger;
import me.doubledutch.stroom.streams.*;
import me.doubledutch.stroom.perf.*;
import me.doubledutch.stroom.*;
import java.util.*;
import me.doubledutch.stroom.client.StreamConnection;
import org.json.*;
import java.net.*;
import javax.script.*;
				JSONObject objSt=obj.getJSONObject("o");
				Iterator<String> keyIt=objSt.keys();

				while(keyIt.hasNext()){
					String key=keyIt.next();
					aggregateMap.put(key,objSt.getLong(key));
				}
			}

			loc+=batch.size();
			batch=getStream("state").get(loc,loc+500);
		}
	}

	private void saveState() throws Exception{
		JSONObject obj=new JSONObject();
		obj.put("i",index);

		// Save snapshot and build state
		List<String> batch=new ArrayList<String>();
		List<String> pKeyList=new ArrayList<String>();
		for(String key:partitionTouchSet){
			pKeyList.add(key);
			String data=aggregateCache.get(key);
			batch.add(data);
		}
		
		metric.startTimer("output.append");
		List<Long> results=null;
		if(batch.size()>0){
// Path: reference/src/main/java/me/doubledutch/stroom/client/StreamConnection.java
// public interface StreamConnection{
// 	public final static int LINEAR=0;
// 	public final static int SYNC=1;
// 	public final static int FLUSH=2;
// 	public final static int NONE=3;
// 
// 	public List<String> get(long index,long endIndex) throws IOException;
// 	public String get(long index) throws IOException;
// 	public String getLast() throws IOException;
// 	public long getCount() throws IOException;
// 	
// 	public long append(JSONObject data) throws IOException,JSONException;
// 	public long append(String data) throws IOException,JSONException;
// 	public List<Long> append(List<String> data) throws IOException,JSONException;
// 	
// 	public long append(JSONObject data,int hint) throws IOException,JSONException;
// 	public long append(String data,int hint) throws IOException,JSONException;
// 	public List<Long> append(List<String> data,int hint) throws IOException,JSONException;
// 
// 	public void truncate(long index) throws IOException;
// }


// Path: reference/src/main/java/me/doubledutch/stroom/aggregates/PartitionedAggregateService.java
import org.apache.log4j.Logger;
import me.doubledutch.stroom.streams.*;
import me.doubledutch.stroom.perf.*;
import me.doubledutch.stroom.*;
import java.util.*;
import me.doubledutch.stroom.client.StreamConnection;
import org.json.*;
import java.net.*;
import javax.script.*;

				JSONObject objSt=obj.getJSONObject("o");
				Iterator<String> keyIt=objSt.keys();

				while(keyIt.hasNext()){
					String key=keyIt.next();
					aggregateMap.put(key,objSt.getLong(key));
				}
			}

			loc+=batch.size();
			batch=getStream("state").get(loc,loc+500);
		}
	}

	private void saveState() throws Exception{
		JSONObject obj=new JSONObject();
		obj.put("i",index);

		// Save snapshot and build state
		List<String> batch=new ArrayList<String>();
		List<String> pKeyList=new ArrayList<String>();
		for(String key:partitionTouchSet){
			pKeyList.add(key);
			String data=aggregateCache.get(key);
			batch.add(data);
		}
		
		metric.startTimer("output.append");
		List<Long> results=null;
		if(batch.size()>0){
			results=getStream("output").append(batch,StreamConnection.FLUSH);