package org.lumongo.server.hazelcast;

import java.io.Serializable;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;

import org.apache.log4j.Logger;

import com.hazelcast.core.Member;

public class UpdateSegmentsTask implements Callable<Void>, Serializable {
	
	private static Logger log = Logger.getLogger(UpdateSegmentsTask.class);
	
	private static final long serialVersionUID = 1L;
	
	private final int hazelcastPort;
	
	private final String indexName;
	private final Map<Member, Set<Integer>> newMemberToSegmentMap;
	
	public UpdateSegmentsTask(int hazelcastPort, String indexName, Map<Member, Set<Integer>> newMemberToSegmentMap) {
		this.hazelcastPort = hazelcastPort;
		this.indexName = indexName;
		this.newMemberToSegmentMap = newMemberToSegmentMap;
	}
	
	@Override
	public Void call() throws Exception {
		try {
			HazelcastManager hazelcastManger = HazelcastManager.getHazelcastManager(hazelcastPort);
			hazelcastManger.updateSegmentMap(indexName, newMemberToSegmentMap);
		}
		catch (Exception e) {
			log.error(e.getClass().getSimpleName() + ": ", e);
			throw e;
		}
		return null;
	}
}

@@ -43,113 +43,219 @@
 ort 
-java.io.Serializable;%0Aimport java.util.Map;%0Aimport java.util.Set;%0Aimport java.util.concurrent.Call
+com.hazelcast.core.Member;%0Aimport com.hazelcast.instance.MemberImpl;%0Aimport com.hazelcast.nio.ObjectDataInput;%0Aimport com.hazelcast.nio.ObjectDataOutput;%0Aimport com.hazelcast.nio.serialization.DataSerializ
 able;%0A
-%0A
 impo
@@ -294,33 +294,161 @@
 ort 
-com.hazelcast.core.Member
+java.io.IOException;%0Aimport java.util.HashMap;%0Aimport java.util.HashSet;%0Aimport java.util.Map;%0Aimport java.util.Set;%0Aimport java.util.concurrent.Callable
 ;%0A%0Ap
@@ -505,16 +505,20 @@
 %3CVoid%3E, 
+Data
 Serializ
@@ -526,16 +526,66 @@
 ble %7B%0A%09%0A
+%09private static final long serialVersionUID = 1L;%0A
 %09private
@@ -649,18 +649,16 @@
 class);%0A
-%09%0A
 %09private
@@ -662,66 +662,8 @@
 ate 
-static final long serialVersionUID = 1L;%0A%09%0A%09private final 
 int 
@@ -688,22 +688,16 @@
 private 
-final 
 String i
@@ -715,22 +715,16 @@
 private 
-final 
 Map%3CMemb
@@ -764,17 +764,52 @@
 entMap;%0A
-%09
+%0A%09public UpdateSegmentsTask() %7B%0A%0A%09%7D%0A
 %0A%09public
@@ -1378,11 +1378,928 @@
 null;%0A%09%7D
+%0A%0A%09@Override%0A%09public void writeData(ObjectDataOutput out) throws IOException %7B%0A%09%09out.writeInt(hazelcastPort);%0A%09%09out.writeUTF(indexName);%0A%09%09out.writeInt(newMemberToSegmentMap.size());%0A%0A%09%09for (Map.Entry%3CMember, Set%3CInteger%3E%3E e : newMemberToSegmentMap.entrySet()) %7B%0A%09%09%09e.getKey().writeData(out);%0A%09%09%09out.writeInt(e.getValue().size());%0A%09%09%09for (Integer segment : e.getValue()) %7B%0A%09%09%09%09out.writeInt(segment);%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%7D%0A%0A%09@Override%0A%09public void readData(ObjectDataInput in) throws IOException %7B%0A%09%09hazelcastPort = in.readInt();%0A%09%09indexName = in.readUTF();%0A%09%09newMemberToSegmentMap = new HashMap%3C%3E();%0A%09%09int count = in.readInt();%0A%0A%09%09for (int i = 0; i %3C count; i++) %7B%0A%09%09%09MemberImpl mi = new MemberImpl();%0A%09%09%09mi.readData(in);%0A%09%09%09Set%3CInteger%3E segments = new HashSet%3C%3E();%0A%09%09%09int segCount = in.readInt();%0A%09%09%09for (int j = 0; j %3C segCount; j++) %7B%0A%09%09%09%09segments.add(in.readInt());%0A%09%09%09%7D%0A%09%09%09newMemberToSegmentMap.put(mi, segments);%0A%09%09%7D%0A%0A%09%7D
 %0A%7D%0A

fix hazelcast serialization of updatesegmentstask