UDP.rar_DataSwapEvent_udp封装论坛

package unicast; import java.io.*; import java.net.*; import java.util.*; public class DataPacket { InputStream is; OutputStream os; PacketHeader header; PacketBody body; ArrayList al; public static final int DataSwapSize = 64532; /** * 在接收数据报使用 */ public DataPacket() { header = new PacketHeader(); body = new PacketBody(); al = new ArrayList(); } /** * 在发送数据报时使用,它调用报文分割操作. * * @param file * String 硬盘文件 */ public DataPacket(String file) { this(); try { is = new FileInputStream(file); header.CalcHeaderInfo(is.available()); this.madeBody(); is.close(); // this.Gereratedata(); } catch (FileNotFoundException ex) { ex.printStackTrace(); } catch (IOException ex1) { ex1.printStackTrace(); } } /** * 在发送数据报时使用,它调用报文分割操作. * * @param url * URL url地址 */ public DataPacket(URL url) { this(); try { // is = url.openStream(); URLConnection conn = url.openConnection(); is = conn.getInputStream(); int total = conn.getContentLength(); header.CalcHeaderInfo(total); this.madeBody(); // System.out.println(total+":"+total); is.close(); } catch (IOException ex) { ex.printStackTrace(); } } /** * 为发送构造分组,使用PackageHeader处理了报头格式,并为分组编序号. */ @SuppressWarnings({ "unchecked", "static-access" }) private void madeBody() { al.clear(); byte[] buffer; DataEntry de; for (int i = 0; i < header.fragmentcounter; i++) { try { ByteArrayOutputStream bos = new ByteArrayOutputStream(); // is.skip(i * body.BODY_BUFFER_SIZE); header.ArrageSort(i); de = new DataEntry(PacketBody.BODY_BUFFER_SIZE); de.setSn(i); de.setStreamsize(header.getStreamsize()); de.setFragmentcounter(header.getFragmentcounter()); if (header.isWTailFragment(i)) { buffer = new byte[header.getMinfragment()]; is.read(buffer, 0, buffer.length); header.setActByteSize(header.getMinfragment()); de.setActByteSize(header.getMinfragment()); } else { buffer = new byte[body.BODY_BUFFER_SIZE]; is.read(buffer, 0, buffer.length); } // System.out.println("length-------"+i+" // "+body.getBody().length+" "+header.getMinfragment()); body.setBody(buffer); // System.out.println("length:" + i + " " + header.toString()); bos.write(header.getByte(), 0, header.HEADER_BUFFER_SIZE); bos.write(body.getBody(), 0, body.getBody().length); de.setBytes(bos.toByteArray()); al.add(de); } catch (IOException ex) { ex.printStackTrace(); } } } /** * 为发送构造分组,没有考虑报头格式,也没有为分组编序号. */ @SuppressWarnings({ "unused", "static-access", "unchecked" }) private void madeBody1() { al.clear(); for (int i = 0; i < header.fragmentcounter; i++) { try { if (header.isWTailFragment(i)) is.read(body.getBody(), i * body.BODY_BUFFER_SIZE, header .getMinfragment()); else is.read(body.getBody(), i * body.BODY_BUFFER_SIZE, body.BODY_BUFFER_SIZE); ByteArrayOutputStream bos = new ByteArrayOutputStream(); bos.write(header.getByte(), 0, header.HEADER_BUFFER_SIZE); bos.write(body.getBody(), header.HEADER_BUFFER_SIZE, body .getBody().length); al.add(bos); } catch (IOException ex) { ex.printStackTrace(); } } } /** * 在接收到报文后,对此报文执行组装,并处理报文丢失和乱序情况. * * @param b1 * byte[] */ @SuppressWarnings("unchecked") public void Add(byte[] b1) { byte[] buffer = (byte[]) b1.clone(); handlerText(buffer); DataEntry de = new DataEntry(buffer, header.getActByteSize()); de.setSn(header.getSn()); de.setStreamsize(header.getStreamsize()); de.setFragmentcounter(header.getFragmentcounter()); al.add(de); } @SuppressWarnings("static-access") private void handlerText(byte[] buffer) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); baos.write(buffer, 0, header.HEADER_BUFFER_SIZE); byte[] b = new byte[header.HEADER_BUFFER_SIZE]; System.arraycopy(buffer, 0, b, 0, b.length); ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray()); InputStreamReader isr = new InputStreamReader(bais); BufferedReader br = new BufferedReader(isr); try { header = new PacketHeader(br.readLine()); } catch (Exception ex) { ex.printStackTrace(); } } @SuppressWarnings("unused") private String calFileSize(int size) { return size / 1024 + "K"; } public ArrayList getDataPackets() { return al; } /** * 是否接收完毕,通过序号是否等于最大段数来判断,这也许有问题,比如,正好是最后一个段丢失了,这样 这个包整个就丢失了. * * @return */ public boolean isFull() { return this.header.getSn() == this.header.getFragmentcounter() - 1 ? true : false; } /** * 判断是否只有一个段. * * @return */ public boolean isZero() { return this.header.getSn() == 0 ? true : false; } /** * 该函数执行报文组装,不考虑丢失的报文. * * @return */ @SuppressWarnings({ "unused", "static-access" }) private ByteArrayOutputStream fetchDataPackets() { ByteArrayOutputStream bos = new ByteArrayOutputStream(); byte[] buffer = null; DataEntry de; for (int i = 0; i < al.size(); i++) { try { de = this.getSnData(i); buffer = de.getByte(); if (header.getStreamsize() == de.getStreamsize()) { bos.write(de.getByte(), header.HEADER_BUFFER_SIZE, de .getActByteSize()); System.out.println(de.toString() + " -- fetchDataPackets"); } } catch (Exception ex) { ex.printStackTrace(); } } return bos; } /** * 该函数执行报文组装,对于丢失的报文,写入空报文. * * @return ByteArrayOutputStream */ @SuppressWarnings("static-access") private ByteArrayOutputStream fetchDataPackets_sn() { ByteArrayOutputStream bos = new ByteArrayOutputStream(); @SuppressWarnings("unused") byte[] buffer; DataEntry de; for (int i = 0; i < header.getFragmentcounter(); i++) { try { de = this.getSnData(i); if (de == null) { de = seachDeData(i); } buffer = de.getByte(); // System.out.println(de.getSn() + ":" + i); // handlerText(buffer); // bos.write(buffer, header.HEADER_BUFFER_SIZE, // buffer.length - header.HEADER_BUFFER_SIZE); if (header.getStreamsize() == de.getStreamsize()) { bos.write(de.getByte(), header.HEADER_BUFFER_SIZE, de .getActByteSize()); // System.out.println(de.toString()); } } catch (Exception ex) { ex.printStackTrace(); } } return bos; } /** * 对缓冲的数据包进行排序处理,即按顺序提取同一帧的数据,如果没有找到该序号的帧,则返回空值. * * @param sn * int 要找的帧序号. * @return DataEntry */ private DataEntry getSnData(int sn) { DataEntry de = null; for (int i = 0; i < al.size(); i++) { de = (DataEntry) al.get(i); if (header.getStreamsize() == de.getStreamsize()) { if (sn == de.getSn()) break; else de = null; } } return de; } /** * 按序号开始向前或者是向后寻找最近的帧片段,日后可以增加请求重发功能,通过开一个通信连接. * * @param sn * int * @return DataEntry */ private DataEntry seachDeData(int sn) { DataEntry de = null; int initvalue, minvalue = 10000; @SuppressWarnings("unused") DataEntry back, fore = null; for (int i = 0; i < al.size(); i++) { de = (DataEntry) al.get(i); if (header.getStreamsize() == de.getStreamsize()) { initvalue = Math.abs(de.getSn() - sn); if (de.getFragmentcounter() != de.getSn() && initvalue < minvalue) { minvalue = initvalue; fore = de; } } } return fore; } /** * 除去最后一帧外,随机抽取一帧. *