From 67e007a758afd636ca13cf54e00c91a786f72103 Mon Sep 17 00:00:00 2001 From: ascrutae Date: Tue, 5 Jul 2016 08:11:40 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E9=83=A8=E5=88=86=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E7=BC=96=E8=AF=91=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cloud/skywalking/buffer/BufferGroup.java | 23 +++-- .../com/ai/cloud/skywalking/conf/Config.java | 2 + .../invoke/monitor/BaseInvokeMonitor.java | 11 ++- .../monitor/RPCClientInvokeMonitor.java | 3 + .../cloud/skywalking/sender/DataSender.java | 4 +- .../sender/DataSenderWithCopies.java | 2 +- .../cloud/skywalking/sender/IDataSender.java | 6 +- .../ai/cloud/skywalking/protocol/AckSpan.java | 9 ++ .../skywalking/protocol/RequestSPan.java | 19 ++++ .../SerializableDataTypeRegister.java | 22 +++++ .../cloud/skywalking/protocol/SpanData.java | 19 ++-- .../common/AbstractDataSerializable.java | 2 +- .../protocol/common/ISerializable.java | 2 +- .../cloud/skywalking/util/IntegerAssist.java | 9 ++ .../skywalking/util/TransportPackager.java | 90 ++++++++++++++----- 15 files changed, 172 insertions(+), 51 deletions(-) diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java index 574fb793c..c20f53999 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/buffer/BufferGroup.java @@ -1,10 +1,8 @@ package com.ai.cloud.skywalking.buffer; import com.ai.cloud.skywalking.conf.Config; -import com.ai.cloud.skywalking.conf.Constants; import com.ai.cloud.skywalking.logging.LogManager; import com.ai.cloud.skywalking.logging.Logger; -import com.ai.cloud.skywalking.protocol.Span; import com.ai.cloud.skywalking.protocol.common.ISerializable; import com.ai.cloud.skywalking.selfexamination.HeathReading; import com.ai.cloud.skywalking.selfexamination.SDKHealthCollector; @@ -45,7 +43,7 @@ public class BufferGroup { } } - public void save(Span span) { + public void save(ISerializable data) { int i = index.getAndIncrement(); if (dataBuffer[i] != null) { logger.warn( @@ -53,7 +51,7 @@ public class BufferGroup { groupName, i); SDKHealthCollector.getCurrentHeathReading("BufferGroup").updateData(HeathReading.WARNING, "BufferGroup index[" + i + "] data collision, data been coverd."); } - dataBuffer[i] = span; + dataBuffer[i] = data; SDKHealthCollector.getCurrentHeathReading("BufferGroup").updateData(HeathReading.INFO, "save span"); } @@ -69,7 +67,6 @@ public class BufferGroup { @Override public void run() { - //TODO: 根据list长度进行限制 List packageData = new ArrayList(); while (true) { boolean bool = false; @@ -79,33 +76,33 @@ public class BufferGroup { continue; } bool = true; - if (data.length() + dataBuffer[i].toString().length() >= Config.Sender.MAX_SEND_LENGTH) { + if (packageData.size() >= Config.Sender.MAX_SEND_DATA_SIZE) { while (!DataSenderFactoryWithBalance.getSender() - .send(data.toString())) { + .send(packageData)) { try { Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL); } catch (InterruptedException e) { logger.error("Sleep Failure"); } } - logger.debug("send buried-point data, size:{}", data.length()); - data = new StringBuilder(); + logger.debug("send buried-point data, size:{}", packageData.size()); + packageData = new ArrayList(); } - data.append(dataBuffer[i] + Constants.DATA_SPILT); + packageData.add(dataBuffer[i]); dataBuffer[i] = null; } - if (data != null && data.length() > 0) { + if (packageData != null && packageData.size() > 0) { while (!DataSenderFactoryWithBalance.getSender().send( - data.toString())) { + packageData)) { try { Thread.sleep(CONSUMER_FAIL_RETRY_WAIT_INTERVAL); } catch (InterruptedException e) { logger.error("Sleep Failure"); } } - data = new StringBuilder(); + packageData = new ArrayList(); } } catch (Throwable e) { logger.error("buffer group running failed", e); diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java index 648d149ce..8a7fa8542 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java @@ -46,6 +46,8 @@ public class Config { } public static class Sender { + // 最大发送数据个数 + public static final int MAX_SEND_DATA_SIZE = 10; // 最大发送者的连接数阀比例 public static int CONNECT_PERCENT = 50; diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java index 46fe999ca..0cf3017e8 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/BaseInvokeMonitor.java @@ -1,5 +1,6 @@ package com.ai.cloud.skywalking.invoke.monitor; +import com.ai.cloud.skywalking.buffer.BufferGroup; import com.ai.cloud.skywalking.buffer.ContextBuffer; import com.ai.cloud.skywalking.conf.AuthDesc; import com.ai.cloud.skywalking.conf.Config; @@ -7,6 +8,8 @@ import com.ai.cloud.skywalking.context.CurrentThreadSpanStack; import com.ai.cloud.skywalking.logging.LogManager; import com.ai.cloud.skywalking.logging.Logger; import com.ai.cloud.skywalking.model.ContextData; +import com.ai.cloud.skywalking.protocol.AckSpan; +import com.ai.cloud.skywalking.protocol.RequestSpan; import com.ai.cloud.skywalking.protocol.Span; import java.util.HashSet; @@ -31,6 +34,8 @@ public abstract class BaseInvokeMonitor { + "\tParentLevelId:" + spanData.getParentLevel() + "\tLevelId:" + spanData.getLevelId()); } + // 根据SpanData生成RequestSpan，并保存 + ContextBuffer.save(new RequestSpan(spanData)); // 将新创建的Context存放到ThreadLocal栈中。 CurrentThreadSpanStack.push(spanData); @@ -58,10 +63,10 @@ public abstract class BaseInvokeMonitor { + "\tviewpointId:" + spanData.getViewPointId() + "\tParentLevelId:" + spanData.getParentLevel() + "\tLevelId:" + spanData.getLevelId() - + "\tbusinessKey:" + spanData.getBusinessKey()); + + "\tbusinessKey:" + spanData.getParameters()); } - - ContextBuffer.save(spanData); + // 生成并保存到缓存 + ContextBuffer.save(new AckSpan(spanData)); } catch (Throwable t) { logger.error(t.getMessage(), t); } diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/RPCClientInvokeMonitor.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/RPCClientInvokeMonitor.java index df9c28414..6dce553ad 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/RPCClientInvokeMonitor.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/invoke/monitor/RPCClientInvokeMonitor.java @@ -1,5 +1,6 @@ package com.ai.cloud.skywalking.invoke.monitor; +import com.ai.cloud.skywalking.buffer.ContextBuffer; import com.ai.cloud.skywalking.conf.AuthDesc; import com.ai.cloud.skywalking.context.CurrentThreadSpanStack; import com.ai.cloud.skywalking.logging.LogManager; @@ -7,6 +8,7 @@ import com.ai.cloud.skywalking.logging.Logger; import com.ai.cloud.skywalking.model.ContextData; import com.ai.cloud.skywalking.model.EmptyContextData; import com.ai.cloud.skywalking.model.Identification; +import com.ai.cloud.skywalking.protocol.RequestSpan; import com.ai.cloud.skywalking.protocol.Span; import com.ai.cloud.skywalking.protocol.common.SpanType; import com.ai.cloud.skywalking.util.ContextGenerator; @@ -25,6 +27,7 @@ public class RPCClientInvokeMonitor extends BaseInvokeMonitor { //设置SpanType的类型 spanData.setSpanType(SpanType.RPC_CLIENT); + ContextBuffer.save(new RequestSpan(spanData)); CurrentThreadSpanStack.push(spanData); return new ContextData(spanData.getTraceId(), generateSubParentLevelId(spanData), spanData.getCallType()); diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java index f1b56f3c3..c8fb497c0 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java @@ -70,7 +70,7 @@ public class DataSender implements IDataSender { /** * 返回是否发送成功 * - * @param data + * @param packageData * @return */ @Override @@ -78,7 +78,7 @@ public class DataSender implements IDataSender { try { if (channel != null && channel.isActive()) { - byte[] dataPackage = TransportPackager.pack(data.getBytes()); + byte[] dataPackage = TransportPackager.pack(packageData); channel.writeAndFlush(dataPackage); SDKHealthCollector.getCurrentHeathReading("sender").updateData(HeathReading.INFO, "DataSender[" + socketAddress + "] send data successfully."); diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java index c562ee002..e9898acbc 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderWithCopies.java @@ -53,7 +53,7 @@ public class DataSenderWithCopies implements IDataSender { public boolean send(List packageData) { int successNum = 0; for (IDataSender sender : senders) { - if (sender.send(data)) { + if (sender.send(packageData)) { successNum++; } } diff --git a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/IDataSender.java b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/IDataSender.java index ebe2b7edf..7e0dd2d1e 100644 --- a/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/IDataSender.java +++ b/skywalking-collector/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/IDataSender.java @@ -1,5 +1,9 @@ package com.ai.cloud.skywalking.sender; +import com.ai.cloud.skywalking.protocol.common.ISerializable; + +import java.util.List; + public interface IDataSender { - public boolean send(String data); + public boolean send(List data); } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java index 26ef3bbdf..16a8c63ca 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/AckSpan.java @@ -37,6 +37,15 @@ public class AckSpan extends AbstractDataSerializable { */ private String exceptionStack; + public AckSpan(Span spanData) { + this.traceId = spanData.getTraceId(); + this.parentLevel = spanData.getParentLevel(); + this.levelId = spanData.getLevelId(); + this.cost = System.currentTimeMillis() - spanData.getStartDate(); + this.statusCode = spanData.getStatusCode(); + this.exceptionStack = spanData.getExceptionStack(); + } + public String getTraceId() { return traceId; } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java index 70b6781e6..7a71d5305 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/RequestSPan.java @@ -78,6 +78,25 @@ public class RequestSpan extends AbstractDataSerializable { private Map paramters = new HashMap(); + public RequestSpan(Span spanData) { + this.traceId = spanData.getTraceId(); + this.parentLevel = spanData.getParentLevel(); + this.levelId = spanData.getLevelId(); + this.address = spanData.getAddress(); + this.applicationId = spanData.getApplicationId(); + this.callType = spanData.getCallType(); + this.spanType = spanData.getSpanType(); + this.spanTypeDesc = spanData.getSpanTypeDesc(); + this.userId = spanData.getUserId(); + if (isEntrySpan(spanData)) { + this.paramters.putAll(spanData.getParamters()); + } + } + + private boolean isEntrySpan(Span spanData) { + return "0".equals(spanData.getParentLevel() + spanData.getLevelId()); + } + public String getTraceId() { return traceId; } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializableDataTypeRegister.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializableDataTypeRegister.java index c7ca8fcb5..363db522a 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializableDataTypeRegister.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SerializableDataTypeRegister.java @@ -1,9 +1,13 @@ package com.ai.cloud.skywalking.protocol; +import com.ai.cloud.skywalking.protocol.common.AbstractDataSerializable; +import com.ai.cloud.skywalking.protocol.common.ISerializable; +import com.ai.cloud.skywalking.protocol.common.NullableClass; import com.ai.cloud.skywalking.protocol.exception.SerializableDataTypeRegisterException; import java.util.HashMap; import java.util.Map; +import java.util.Set; /** * Created by wusheng on 16/7/4. @@ -33,6 +37,24 @@ public class SerializableDataTypeRegister { } } + public static ISerializable findSerializableClassAndSerialize(byte[] data) { + Set> registerSerializableClasses = CLASS_MAPPING_DATA_TYPE.keySet(); + for (Class serializableClass : registerSerializableClasses) { + try { + ISerializable result = ((ISerializable) serializableClass.newInstance()); + //TODO + NullableClass nullableClass = result.convert2Object(data); + if (!nullableClass.isNull()){ + + } + } catch (Exception e) { + + } + + } + return null; + } + public static Integer getType(Class clazz) { if (CLASS_MAPPING_DATA_TYPE.containsKey(clazz)) { return CLASS_MAPPING_DATA_TYPE.get(clazz); diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SpanData.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SpanData.java index 48eaced28..2ea223703 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SpanData.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/SpanData.java @@ -3,6 +3,9 @@ package com.ai.cloud.skywalking.protocol; import com.ai.cloud.skywalking.protocol.common.CallType; import com.ai.cloud.skywalking.protocol.common.SpanType; +import java.util.HashMap; +import java.util.Map; + public abstract class SpanData { /** * Span在序列中，各字段间的分隔符 @@ -92,7 +95,7 @@ public abstract class SpanData { * 节点调用过程中的业务字段
* 如：业务系统设置的订单号，SQL语句等 */ - protected String businessKey = ""; + protected Map parameters = new HashMap(); /** * 节点调用的所在进程号 */ @@ -177,10 +180,6 @@ public abstract class SpanData { return isInvalidate; } - public void setBusinessKey(String businessKey) { - this.businessKey = businessKey; - } - public void setProcessNo(String processNo) { this.processNo = processNo; } @@ -209,8 +208,12 @@ public abstract class SpanData { this.exceptionStack = exceptionStack; } - public String getBusinessKey() { - return businessKey; + public Map getParameters() { + return parameters; + } + + public void setParameters(Map parameters) { + this.parameters = parameters; } public String getProcessNo() { @@ -232,4 +235,6 @@ public abstract class SpanData { public String getCallType() { return callType; } + + } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java index 619d0bab8..2a3025ce2 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/AbstractDataSerializable.java @@ -37,7 +37,7 @@ public abstract class AbstractDataSerializable implements ISerializable, Nullabl } @Override - public Object convert2Object(byte[] data) { + public NullableClass convert2Object(byte[] data) { // TODO:data的前4位转成type; int dataType = 1; if(!SerializableDataTypeRegister.isTypeAndClassMatch(dataType, this.getClass())){ diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java index 06712bb75..81ce0d536 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/protocol/common/ISerializable.java @@ -6,6 +6,6 @@ package com.ai.cloud.skywalking.protocol.common; public interface ISerializable { byte[] convert2Bytes(); - Object convert2Object(byte[] data); + NullableClass convert2Object(byte[] data); } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/IntegerAssist.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/IntegerAssist.java index 76c18155e..8e2309ec9 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/IntegerAssist.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/IntegerAssist.java @@ -12,4 +12,13 @@ public class IntegerAssist { src[3] = (byte) (value & 0xFF); return src; } + + public static int bytesToInt(byte[] src, int offset) { + int value; + value = (int) (((src[offset] & 0xFF) << 24) + | ((src[offset + 1] & 0xFF) << 16) + | ((src[offset + 2] & 0xFF) << 8) + | (src[offset + 3] & 0xFF)); + return value; + } } diff --git a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java index 2a9ad6861..c013283ea 100644 --- a/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java +++ b/skywalking-collector/skywalking-protocol/src/main/java/com/ai/cloud/skywalking/util/TransportPackager.java @@ -1,48 +1,52 @@ package com.ai.cloud.skywalking.util; +import com.ai.cloud.skywalking.protocol.SerializableDataTypeRegister; import com.ai.cloud.skywalking.protocol.common.ISerializable; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; public class TransportPackager { - public static byte[] pack(List packageData) { + public static byte[] pack(List beSendingData) { // 对协议格式进行修改 // | check sum(4 byte) | data - byte[] dataPackage = new byte[data.length + 4]; - - packDataText(data, dataPackage); - packCheckSum(data, dataPackage); - + byte[] dataText = packDataText(beSendingData); + byte[] dataPackage = packCheckSum(dataText); return dataPackage; } - private static void packCheckSum(byte[] data, byte[] dataPackage) { - byte[] checkSumArray = generateChecksum(data, 0); - System.arraycopy(checkSumArray, 0, dataPackage, 0, 4); - } - - private static void packDataText(byte[] data, byte[] dataPackage) { - System.arraycopy(data, 0, dataPackage, 4, data.length); - } - - public static List unpack(byte[] dataPackage) { if (validateCheckSum(dataPackage)) { return unpackDataText(dataPackage); } else { return null; } - } - private static byte[] unpackDataText(byte[] dataPackage) { - byte[] data = new byte[dataPackage.length - 4]; - System.arraycopy(dataPackage, 4, data, 0, data.length); - return data; + private static List unpackDataText(byte[] dataPackage) { + List serializeData = new ArrayList(); + int currentLength = 0; + while (true) { + //读取长度 + int dataLength = IntegerAssist.bytesToInt(dataPackage, 0); + // 反序列化 + byte[] data = new byte[dataLength]; + System.arraycopy(dataPackage, currentLength + 4, data, 0, dataLength); + // + ISerializable data1 = SerializableDataTypeRegister.findSerializableClassAndSerialize(data); + if (data1 != null) { + serializeData.add(data1); + } + currentLength = 4 + dataLength; + if (currentLength >= dataPackage.length) { + break; + } + } + return serializeData; } - private static boolean validateCheckSum(byte[] dataPackage){ + private static boolean validateCheckSum(byte[] dataPackage) { byte[] checkSum = generateChecksum(dataPackage, 4); byte[] originCheckSum = new byte[4]; System.arraycopy(dataPackage, 0, originCheckSum, 0, 4); @@ -63,4 +67,46 @@ public class TransportPackager { return IntegerAssist.intToBytes(result); } + + private static byte[] packCheckSum(byte[] dataText) { + byte[] dataPackage = new byte[dataText.length + 4]; + byte[] checkSum = generateChecksum(dataText, 0); + System.arraycopy(checkSum, 0, dataPackage, 0, 4); + System.arraycopy(dataText, 0, dataPackage, 4, dataText.length); + return dataPackage; + } + + private static byte[] packDataText(List beSendingData) { + byte[] dataText = new byte[1024 * 30]; + int currentIndex = 0; + for (ISerializable sendingData : beSendingData) { + byte[] dataElementText = appendingLength(sendingData.convert2Bytes()); + dataText = expansionCapacityIfNecessary(dataText, currentIndex, dataElementText); + appendBeSendingDataText(dataElementText, dataText, currentIndex); + currentIndex += dataElementText.length; + } + + return dataText; + } + + private static void appendBeSendingDataText(byte[] beSendingDataText, byte[] dataText, int currentIndex) { + System.arraycopy(beSendingDataText, 0, dataText, currentIndex, beSendingDataText.length); + } + + private static byte[] expansionCapacityIfNecessary(byte[] dataText, int currentLength, byte[] beSendingDataText) { + if (beSendingDataText.length + currentLength > 1024 * 30) { + byte[] newDataText = new byte[dataText.length + 1024 * 30]; + System.arraycopy(dataText, 0, newDataText, 0, dataText.length); + return newDataText; + } + return dataText; + } + + private static byte[] appendingLength(byte[] dataByte) { + byte[] dataText = new byte[dataByte.length + 4]; + System.arraycopy(dataByte, 0, dataText, 4, dataByte.length); + byte[] length = IntegerAssist.intToBytes(dataByte.length); + System.arraycopy(length, 0, dataText, 0, 4); + return dataText; + } }