After running for a period of time, the receiver cannot receive data

[chenqinggang001]

I tried to use norm to send data stream under android, it works fine, but whenever it runs continuously for a period of time, it cannot receive data, abnormal content:
NormStreamObject::WriteSegment() broken stream data not read by app!

my code:

public class NormStreamRecv {
	static final long SESSION_BUFFER_SIZE = 1024 * 1024;
	static final int SEGMENT_SIZE = 1400;
	static final int BLOCK_SIZE = 64;
	static final int PARITY_SEGMENTS = 16;
	static final String DEST_ADDRESS = "239.244.0.100";
	static final int DEST_PORT = 10030;

	public static void start() {
		NormInstance instance = null;
		NormSession session = null;
		String destAddress = DEST_ADDRESS;
		int destPort = DEST_PORT;

		try {
			int length = 0;
			int offset = 0;
			byte[] buf = new byte[65536];
			boolean useUnicastNACKs = false;
			instance = new NormInstance();
			session = instance.createSession(destAddress, destPort,
					NormNode.NORM_NODE_ANY);
			session.setDefaultUnicastNack(useUnicastNACKs);
			session.startReceiver(SESSION_BUFFER_SIZE);
			boolean streamIsAlive = true;
			NormEvent event;

			Log.d("NormStreamRecv", "Starting receiver...");
			while ((null != (event = instance.getNextEvent())) && streamIsAlive) {
				NormEventType eventType = event.getType();
				NormObject normObject = event.getObject();
				switch (eventType) {
					case NORM_RX_OBJECT_NEW:
						Log.d("NormStreamRecv", "New stream!");
						break;

					case NORM_RX_OBJECT_UPDATED:
						if (normObject instanceof NormStream) {
							int numRead = 0;
							NormStream normStreamobj;
							normStreamobj = (NormStream)normObject;

							while (0 < (numRead = normStreamobj.read(buf, 0, buf.length))) {
								if (-1 != numRead) {
									Log.d("NormStreamRecv", "Read: " + numRead);
								}
							}
						} else {
							Log.d("NormStreamRecv", "Unknown object type!");
						}
						break;

					case NORM_RX_OBJECT_COMPLETED:
						Log.d("NormStreamRecv", "Completed stream!");
						streamIsAlive = false;
						break;
				}
			}
		}
		catch (IOException ex) {
			ex.printStackTrace();
		}
		catch (NumberFormatException ex) {
			Log.d("NormStreamRecv", "NumberFormatException: " + DEST_ADDRESS + ":"+ DEST_PORT + " " + ex.getMessage());
		}
		if (null != session) {
			session.stopReceiver();
			session.destroySession();
		}

		if (null != instance) {
			instance.destroyInstance();
		}

	}
}

public static void start() {
		NormInstance instance = null;
		NormSession session = null;
		NormStream stream = null;
		String destAddress = DEST_ADDRESS;
		int destPort = DEST_PORT;

		try {
			int length = 0;
			int offset = 0;
			byte[] buf = new byte[65536];
			instance = new NormInstance();
			session = instance.createSession(destAddress, destPort,
											 NormNode.NORM_NODE_ANY);
			
			session.setCongestionControl(true, true);

			session.startSender(1, SESSION_BUFFER_SIZE, 
								SEGMENT_SIZE, (short) BLOCK_SIZE, (short) PARITY_SEGMENTS);
			stream = session.streamOpen(REPAIR_WINDOW_SIZE);
			String message = generateRandomString(10000);
			buf = message.getBytes();
			length = buf.length;
			while (true) {
				stream.write(buf, offset, length);
				NormEvent event = instance.getNextEvent();
				NormEventType eventType = event.getType();
				while ((eventType != NormEventType.NORM_TX_QUEUE_EMPTY) &&
						(eventType != NormEventType.NORM_TX_QUEUE_VACANCY)) {
					event = instance.getNextEvent();
					eventType = event.getType();
				}
			}

//			stream.markEom();
//			stream.flush();

		}
		catch (IOException ex) {
			ex.printStackTrace();
		}
		catch (NumberFormatException ex) {
			Log.d("NormStreamSend", "NumberFormatException: " + DEST_ADDRESS + ":"+ DEST_PORT + " " + ex.getMessage());
		}

		if (null != stream) {
			Log.d("NormStreamSend", "Closing stream");
			stream.close(true);
		}

		if (null != session) {
			Log.d("NormStreamSend", "Stopping sender");
			session.stopSender();
			session.destroySession();
		}

		if (null != instance) {
			Log.d("NormStreamSend", "Destroying instance");
			instance.destroyInstance();
		}
	}

	public static String generateRandomString(int length) {
		Random random = new Random();
		StringBuilder sb = new StringBuilder(length);
		String charSet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" +
				"abcdefghijklmnopqrstuvwxyz" + 
				"0123456789";           

		for (int i = 0; i < length; i++) {
			int randomIndex = random.nextInt(charSet.length());
			sb.append(charSet.charAt(randomIndex));
		}
		return sb.toString();
	}

[chenqinggang001]

log
log.txt