import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public FileFragmentProto serialize(FileFragment fragment) {
  FileFragmentProto.Builder builder = FileFragmentProto.newBuilder();
  builder.setId(fragment.inputSourceId);
  builder.setStartOffset(fragment.startKey);
  builder.setLength(fragment.length);
  builder.setPath(fragment.getPath().toString());
  if(fragment.getDiskIds() != null) {
    List<Integer> idList = new ArrayList<>();
    for(int eachId: fragment.getDiskIds()) {
      idList.add(eachId);
    }
    builder.addAllDiskIds(idList);
  }

  if(fragment.hostNames != null) {
    builder.addAllHosts(fragment.hostNames);
  }
  return builder.build();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
@Override
public void readRecord(long fromInstanceId, long toInstanceId, ReadRecordCallback<Builder<? extends Builder>> readCallback) throws IOException,
		UnsupportedChecksumAlgorithm, FileDamageException {
	long tempInstanceId = fromInstanceId;
	DataChunk endChunk = null;
	while (true) {
		DataChunk chunk = fileIndexer.findDataChunk(tempInstanceId);
		if (chunk != null) {
			ReadResult r = chunk.readRecord(tempInstanceId, toInstanceId, readCallback);
			tempInstanceId = r.getMaxSuccessInstanceId() + 1;
			if (chunk == endChunk) {
				break;
			}
			endChunk = chunk;
		} else {
			break;
		}
	}
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder<?> newInstance(byte msgId) {
	Method builder = builders.get(Integer.valueOf(msgId));
	if (builder == null) return null;
	try {
		return (Builder<?>) builder.invoke(null);
		
	} catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
		throw new RuntimeException ("Unable to create new builder instance for msgId " + msgId, e);
	}
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public void read(long instanceId, Builder<? extends Builder> b) {
	if (b instanceof SuccessfulRecord.Builder) {
		sortedReadCallback.read(instanceId, (SuccessfulRecord.Builder)b);
	} else {
		try {
			out.writeVoteRecord(instanceId, (InstanceVoteRecord) b.build(), null);
		}catch (Throwable e) {
			logger.error("upgrade error", e);
			isError = true;
		}
	}
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
public ReadResult initRead(long startInstanceId, ReadRecordCallback<Builder<? extends Builder>> readCallback) throws IOException,
		UnsupportedChecksumAlgorithm {
	if (logger.isDebugEnabled()) {
		logger.debug(String.format("initRead from:%s file:%s", startInstanceId, file.getAbsoluteFile()));
	}
	ReadResult r = readRecord(startInstanceId, Long.MAX_VALUE, readCallback);
	used = r.pos;
	if (this.maxVoteInstanceId < r.maxVoteInstanceId) {
		this.maxVoteInstanceId = r.maxVoteInstanceId;
	}
	this.successfullInstanceId = r.maxSuccessInstanceId;
	return r;
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
public void writeSuccess(long instanceId, SuccessfulRecord.Builder successRecord, Record record,Callable<Object> realEvent) throws IOException, ChunkFullException {

		if (successfullInstanceId >= instanceId) {
			//the success has written and give up write,
			return;
		} else if (successfullInstanceId + 1 == instanceId) {
			checkBufferout();
			int recordLen = record.getSerializeSize();
			if (maxVoteInstanceId >= instanceId || capacity >= (used + recordLen)) {
				record.writeToStream(writeStream, realEvent, realEvent != null);
				++successfullInstanceId;
				if (successRecord.getV().getType() == ValueType.PLACE.getValue()) {
					long value = LongUtil.toLong(successRecord.getV().getValues().toByteArray(), 0);
					if (value > successfullInstanceId) {
						successfullInstanceId = value;
					}
				}
				used += recordLen;
			} else {
				throw new ChunkFullException();
			}
		} else {
			throw new RuntimeException(String.format("excepted successful instanceId %s , give instanceId %s", successfullInstanceId + 1, instanceId));
		}
		if (logger.isDebugEnabled()) {
			logger.debug("instanceId:" + instanceId + " isPlace " + (successRecord.getV().getType() == ValueType.PLACE.getValue()) + " successfullInstanceId "
					+ successfullInstanceId);
		}
	}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
public InstanceSaveContext(long instanceId, cn.com.sparkle.firefly.stablestorage.model.StoreModel.SuccessfulRecord.Builder successfulRecord,
		LinkedList<AddRequestPackage> addRequestPackages) {
	super();
	this.instanceId = instanceId;
	this.successfulRecord = successfulRecord;
	this.addRequestPackages = addRequestPackages;
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
/**
 * 
 * @param obj
 * @param protobuf
 * @return
 */
public static <X extends Builder<X>> X toProtobuf(Object obj, X protobuf) {

	if (null == obj || null == protobuf) {
		return null;
	}

	List<Method> methods = WizardReflexUtil.getGetMethods(obj);

	for (Method method : methods) {
		setProtobuf(obj, protobuf, method);
	}
	return protobuf;
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
/**
 * 
 * @param protobuf
 * @param obj
 * @return
 */
public static <X> X toObject(Builder<?> protobuf, X obj) {

	if (null == protobuf || null == obj) {
		return null;
	}

	Set<FieldDescriptor> fieldSet = protobuf.getAllFields().keySet();

	for (FieldDescriptor field : fieldSet) {
		setObject(protobuf, obj, field);
	}

	return obj;
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
/** @return a new message builder for this messageId, or null if the messageId is unknown */
public Builder<?> newInstance(byte messageId);
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder<?> newInstance(byte messageId) {
  throw new UnsupportedOperationException();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder newBuilder() {
  return ExampleHttpFileFragmentProto.newBuilder();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder newBuilder() {
  return FileFragmentProto.newBuilder();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder newBuilder() {
  return JdbcFragmentProto.newBuilder();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder newBuilder() {
  return KafkaFragmentProto.newBuilder();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public Builder newBuilder() {
  return HBaseFragmentProto.newBuilder();
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
public SplitReader(RecordFileOperator out, long exceptedNextId) {
	this.out = out;
	successReader = new SuccessReader(out);
	sortedReadCallback = new SortedReadCallback<SuccessfulRecord.Builder>(successReader, exceptedNextId);
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
public SuccessfulRecord.Builder getSuccessfulRecord() {
	return successfulRecord;
}
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
public boolean writeSuccessfulRecord(long instanceId, SuccessfulRecord.Builder successfulRecord, LinkedList<AddRequestPackage> addRequestPackages,
final Callable<Object> realEvent) throws IOException, UnsupportedChecksumAlgorithm;
 

import com.google.protobuf.GeneratedMessage.Builder; //导入依赖的package包/类
@Override
public final void read(long instanceId, Builder<? extends Builder> b) {
	if (b instanceof SuccessfulRecord.Builder) {
		readSuccess(instanceId, (SuccessfulRecord.Builder) b);
	}
}