From bb6d7e00e3865c8b95ba5be74d48aed897723f2e Mon Sep 17 00:00:00 2001
From: Leo Ma <begeekmyfriend@gmail.com>
Date: Thu, 14 Apr 2016 19:07:27 +0800
Subject: [PATCH] Sleep in RTMP reading thread

Signed-off-by: Leo Ma <begeekmyfriend@gmail.com>
---
 .../java/net/ossrs/sea/rtmp/io/ReadThread.java     | 14 +++++++++++---
 .../java/net/ossrs/sea/rtmp/io/RtmpConnection.java |  7 +++++--
 .../java/net/ossrs/sea/rtmp/io/WriteThread.java    |  7 ++++---
 3 files changed, 20 insertions(+), 8 deletions(-)

diff --git a/app/src/main/java/net/ossrs/sea/rtmp/io/ReadThread.java b/app/src/main/java/net/ossrs/sea/rtmp/io/ReadThread.java
index 03e0413..11db711 100644
--- a/app/src/main/java/net/ossrs/sea/rtmp/io/ReadThread.java
+++ b/app/src/main/java/net/ossrs/sea/rtmp/io/ReadThread.java
@@ -1,6 +1,7 @@
 package net.ossrs.sea.rtmp.io;
 
 import java.io.EOFException;
+import java.io.IOException;
 import java.io.InputStream;
 import android.util.Log;
 import net.ossrs.sea.rtmp.packets.RtmpPacket;
@@ -29,13 +30,18 @@ public class ReadThread extends Thread {
 
     @Override
     public void run() {
+        boolean isEof = false;
+
         while (!Thread.interrupted()) {
             try {
                 RtmpPacket rtmpPacket = rtmpDecoder.readPacket(in);
                 packetRxHandler.handleRxPacket(rtmpPacket);
+                if (isEof) {
+                    isEof = false;
+                    Thread.sleep(500);
+                }
             } catch (EOFException eof) {
-                // The handler thread will wait until be invoked.
-                packetRxHandler.handleRxPacket(null);
+                isEof = true;
 //            } catch (WindowAckRequired war) {
 //                Log.i(TAG, "Window Acknowledgment required, notifying packet handler...");
 //                packetRxHandler.notifyWindowAckRequired(war.getBytesRead());
@@ -43,11 +49,13 @@ public class ReadThread extends Thread {
 //                    // Pass to handler
 //                    packetRxHandler.handleRxPacket(war.getRtmpPacket());
 //                }
-            } catch (Exception ex) {
+            } catch (IOException ex) {
                 if (!this.isInterrupted()) {
                     Log.e(TAG, "Caught exception while reading/decoding packet, shutting down...", ex);
                     this.interrupt();
                 }
+            } catch (InterruptedException ie) {
+                ie.printStackTrace();
             }
         }
         // Close inputstream
diff --git a/app/src/main/java/net/ossrs/sea/rtmp/io/RtmpConnection.java b/app/src/main/java/net/ossrs/sea/rtmp/io/RtmpConnection.java
index 0508a86..ce48de1 100644
--- a/app/src/main/java/net/ossrs/sea/rtmp/io/RtmpConnection.java
+++ b/app/src/main/java/net/ossrs/sea/rtmp/io/RtmpConnection.java
@@ -289,8 +289,11 @@ public class RtmpConnection implements RtmpPublisher, PacketRxHandler, ThreadCon
                     }
                     case WINDOW_ACKNOWLEDGEMENT_SIZE:
                         WindowAckSize windowAckSize = (WindowAckSize) rtmpPacket;
-                        Log.d(TAG, "handleRxPacketLoop(): Setting acknowledgement window size to: " + windowAckSize.getAcknowledgementWindowSize());
-                        rtmpSessionInfo.setAcknowledgmentWindowSize(windowAckSize.getAcknowledgementWindowSize());
+                        int size = windowAckSize.getAcknowledgementWindowSize();
+                        Log.d(TAG, "handleRxPacketLoop(): Setting acknowledgement window size: " + size);
+                        rtmpSessionInfo.setAcknowledgmentWindowSize(size);
+                        // Set socket option
+                        socket.setSendBufferSize(size);
                         break;
                     case SET_PEER_BANDWIDTH:
                         int acknowledgementWindowsize = rtmpSessionInfo.getAcknowledgementWindowSize();
diff --git a/app/src/main/java/net/ossrs/sea/rtmp/io/WriteThread.java b/app/src/main/java/net/ossrs/sea/rtmp/io/WriteThread.java
index b3857fd..80248a9 100644
--- a/app/src/main/java/net/ossrs/sea/rtmp/io/WriteThread.java
+++ b/app/src/main/java/net/ossrs/sea/rtmp/io/WriteThread.java
@@ -49,16 +49,17 @@ public class WriteThread extends Thread {
                 }
                 out.flush();
             } catch (SocketException se) {
-                Log.e(TAG, "Caught SocketException during write loop, shutting down", se);
+                Log.e(TAG, "WriteThread: Caught SocketException during write loop, shutting down", se);
                 active = false;
                 continue;
             } catch (IOException ex) {
-                Log.e(TAG, "Caught IOException during write loop, shutting down", ex);
+                Log.e(TAG, "WriteThread: Caught IOException during write loop, shutting down", ex);
                 active = false;
                 continue;  // Exit this thread
             }
 
             // Waiting for next packet
+            Log.d(TAG, "WriteThread: waiting...");
             synchronized (txPacketLock) {
                 try {
                     txPacketLock.wait();
@@ -72,7 +73,7 @@ public class WriteThread extends Thread {
         try {
             out.close();
         } catch (Exception ex) {
-            Log.w(TAG, "Failed to close outputstream", ex);
+            Log.w(TAG, "WriteThread: Failed to close outputstream", ex);
         }
         Log.d(TAG, "exiting");
         if (threadController != null) {