修改face收发包接口,明显减少发包阻塞时间(收包似乎阻塞如旧)

This commit is contained in:
free will committed 2021-10-26 22:28:30 +08:00
1 parent 2a30a11498
commit b6e54b34c4
3 files changed
+79 -31

No files matched your search

+5
View File
@@ -126,6 +126,11 @@ public class LinkService {
return minPacket;
}
public LpPacket receiveQuickPacket() throws LogicFaceException {
LpPacket lpPacket=this.transport.receive();
return lpPacket;
}
/**
* 发送一个lp包分片
* @param buf 分片的数据
+17
View File
@@ -243,6 +243,23 @@ public class LogicFace {
}
}
/**
* 直接读取LpPacket,减少读包阻塞时间
* @param timeout
* @return
* @throws LogicFaceException
*/
public LpPacket receiveQuickCPacketV3(long timeout) throws LogicFaceException {
try {
if(!this.transport.setReadTimeout(timeout)){
return null;
}
return this.linkService.receiveQuickPacket();
} catch (Exception e) {
throw new LogicFaceException("LogicFace.receiveQuickCPacketV3: "+e.getMessage());
}
}
/**
* 从 LogicFace 中接收一个普通推送式网络包, 如果收到的包不是普通推送式网络包,则这个包会被忽略
* @param timeout 读超时时间, 毫秒为单位, 小于 0 表示不超时
+57 -31
View File
@@ -11,51 +11,76 @@ import encoding.*;
* @Copyright: MIN-Group;国家重大科技基础设施——未来网络北大实验室;深圳市信息论与未来网络重点实验室
*/
public class LpPacket implements IEncodingAble {
public LpPacketHeader lpPacketHeader=new LpPacketHeader();
public Payload payload=new Payload();
public LpPacketHeader lpPacketHeader = new LpPacketHeader();
public Payload payload = new Payload();
private byte[] rawPayload = null; // LpPacket转成CPacket,再解码出来的payload字段
public LpPacket(){
public byte[] getRawPayload() {
return rawPayload;
}
public LpPacket(LpPacketHeader lpPacketHeader,Payload payload){
this.lpPacketHeader=lpPacketHeader;
this.payload=payload;
public boolean decodeSelf() {
try {
byte[] payload = this.payload.getValue();
Block block = new Block(payload, true);
MINPacket minPacket = new MINPacket();
if (!minPacket.wireDecode(block)) {
return false;
}
CPacket cPacket=new CPacket().createCPacketByMINPacket(minPacket);
this.rawPayload=cPacket.payload.getValue();
return true;
} catch (BlockException | PacketException e) {
e.printStackTrace();
return false;
}
}
public void setId(long fragmentId){
public LpPacket() {
}
public LpPacket(LpPacketHeader lpPacketHeader, Payload payload) {
this.lpPacketHeader = lpPacketHeader;
this.payload = payload;
}
public void setId(long fragmentId) {
this.lpPacketHeader.setLpPacketFragmentId(new LpPacketFragmentId(fragmentId));
}
public long getId(){
public long getId() {
return this.lpPacketHeader.getLpPacketFragmentId().getId();
}
public void setFragmentNum(long fragmentNum){
public void setFragmentNum(long fragmentNum) {
this.lpPacketHeader.setLpPacketFragmentNum(new LpPacketFragmentNum(fragmentNum));
}
public long getFragmentNum(){
public long getFragmentNum() {
return this.lpPacketHeader.getLpPacketFragmentNum().getFragmentNum();
}
public void setFragmentSeq(long fragmentSeq){
public void setFragmentSeq(long fragmentSeq) {
this.lpPacketHeader.setLpPacketFragmentSeq(new LpPacketFragmentSeq(fragmentSeq));
}
public long getFragmentSeq(){
public long getFragmentSeq() {
return this.lpPacketHeader.getLpPacketFragmentSeq().getFragmentSeq();
}
public void setValue(byte[] value){
public void setValue(byte[] value) {
this.payload.setValue(value);
}
public byte[] getValue(){
public byte[] getValue() {
return this.payload.getValue();
}
/**
* 将 LpPacket 线速编码为一个 TLV
*
* @param encoder
* @return
* @throws ComponentException
@@ -68,39 +93,40 @@ public class LpPacket implements IEncodingAble {
// 编码 TLV-VALUE
// LpHeader
int tmpLen = this.lpPacketHeader.wireEncode(encoder);
if(tmpLen<0){
if (tmpLen < 0) {
return -1;
}
totalLength+=tmpLen;
totalLength += tmpLen;
// Fragment
tmpLen = this.payload.wireEncode(encoder);
if(tmpLen<0){
if (tmpLen < 0) {
return -1;
}
totalLength+=tmpLen;
totalLength += tmpLen;
// 编码 TLV-LENGTH
tmpLen = encoder.prependVarNumber(new VlInt(totalLength));
if(tmpLen<0){
if (tmpLen < 0) {
return -1;
}
totalLength+=tmpLen;
totalLength += tmpLen;
// 编码 TLV-TYPE
tmpLen = encoder.prependVarNumber(new VlInt(TLV.TlvLpPacket));
if(tmpLen<0){
if (tmpLen < 0) {
return -1;
}
totalLength+=tmpLen;
totalLength += tmpLen;
return totalLength;
} catch (ComponentException | EncoderException e) {
throw new PacketException("LpPacket.wireEncode: "+e.getMessage());
throw new PacketException("LpPacket.wireEncode: " + e.getMessage());
}
}
/**
* 从 Block TLV 中解码出一个 LpPacket
*
* @param block
* @return
* @throws ComponentException
@@ -115,22 +141,22 @@ public class LpPacket implements IEncodingAble {
}
// 解析子组件
if(!block.parseSubElements()){
if (!block.parseSubElements()) {
return false;
}
ElementContainer elementContainer=block.getSubElements();
ElementContainer elementContainer = block.getSubElements();
for (int i = 0; i < elementContainer.length(); i++) {
Block newBlock=elementContainer.getBlock(i);
int type=newBlock.getType().getVlIntValue2Int();
switch (type){
Block newBlock = elementContainer.getBlock(i);
int type = newBlock.getType().getVlIntValue2Int();
switch (type) {
case TLV.TlvLpPacketHeader:
if(!this.lpPacketHeader.wireDecode(newBlock)){
if (!this.lpPacketHeader.wireDecode(newBlock)) {
return false;
}
break;
case TLV.TlvPayload:
if(!this.payload.wireDecode(newBlock)){
if (!this.payload.wireDecode(newBlock)) {
return false;
}
break;
@@ -139,7 +165,7 @@ public class LpPacket implements IEncodingAble {
return true;
} catch (VlIntException | BlockException | ComponentException e) {
throw new PacketException("LpPacket.wireDecode: "+e.getMessage());
throw new PacketException("LpPacket.wireDecode: " + e.getMessage());
}
}
}