Merge pull request #76 from ittiam-systems/bug-67

Add a connection timeout to the socket

rtmp-client/src/main/cpp/librtmp-jni.c

@@ -30,7 +30,8 @@ Java_net_butterflytv_rtmp_1client_RtmpClient_nativeAlloc(JNIEnv* env, jobject th
  */
 JNIEXPORT jint JNICALL
 Java_net_butterflytv_rtmp_1client_RtmpClient_nativeOpen(JNIEnv* env, jobject thiz, jstring url_,
-                                                        jboolean isPublishMode, jlong rtmpPointer) {
+                                                        jboolean isPublishMode, jlong rtmpPointer,
+                                                        jint sendTimeoutInMs, jint receiveTimeoutInMs) {
 
     const char *url = (*env)->GetStringUTFChars(env, url_, NULL);
     RTMP *rtmp = (RTMP *) rtmpPointer;
@@ -40,6 +41,8 @@ Java_net_butterflytv_rtmp_1client_RtmpClient_nativeOpen(JNIEnv* env, jobject thi
     }
 
     RTMP_Init(rtmp);
+    rtmp->Link.receiveTimeoutInMs = receiveTimeoutInMs;
+    rtmp->Link.sendTimeoutInMs = sendTimeoutInMs;
     int ret = RTMP_SetupURL(rtmp, url);
 
     if (!ret) {

rtmp-client/src/main/cpp/librtmp/rtmp.c

@@ -338,8 +338,7 @@ RTMP_Init(RTMP *r)
   r->m_nServerBW = 2500000;
   r->m_fAudioCodecs = 3191.0;
   r->m_fVideoCodecs = 252.0;
-  //making timeout value to 10 from 30
-  r->Link.timeout = 10;
+  r->Link.receiveTimeoutInMs = 10000;
   r->Link.swfAge = 30;
 }
 
@@ -445,7 +444,7 @@ RTMP_SetupStream(RTMP *r,
 		 AVal *subscribepath,
 		 AVal *usherToken,
 		 int dStart,
-		 int dStop, int bLiveStream, long int timeout)
+		 int dStop, int bLiveStream, long int timeoutInMs)
 {
   RTMP_Log(RTMP_LOGDEBUG, "Protocol : %s", RTMPProtocolStrings[protocol&7]);
   RTMP_Log(RTMP_LOGDEBUG, "Hostname : %.*s", host->av_len, host->av_val);
@@ -474,7 +473,7 @@ RTMP_SetupStream(RTMP *r,
     RTMP_Log(RTMP_LOGDEBUG, "StopTime      : %d msec", dStop);
 
   RTMP_Log(RTMP_LOGDEBUG, "live     : %s", bLiveStream ? "yes" : "no");
-  RTMP_Log(RTMP_LOGDEBUG, "timeout  : %ld sec", timeout);
+  RTMP_Log(RTMP_LOGDEBUG, "timeoutInMs  : %ld sec", timeoutInMs);
 
 #ifdef CRYPTO
   if (swfSHA256Hash != NULL && swfSize > 0)
@@ -518,7 +517,7 @@ RTMP_SetupStream(RTMP *r,
   r->Link.stopTime = dStop;
   if (bLiveStream)
     r->Link.lFlags |= RTMP_LF_LIVE;
-  r->Link.timeout = timeout;
+  r->Link.receiveTimeoutInMs = timeoutInMs;
 
   r->Link.protocol = protocol;
   r->Link.hostname = *host;
@@ -585,7 +584,7 @@ static struct urlopt {
   	"Stream stop position in milliseconds" },
   { AVC("buffer"),    OFF(m_nBufferMS),        OPT_INT, 0,
   	"Buffer time in milliseconds" },
-  { AVC("timeout"),   OFF(Link.timeout),       OPT_INT, 0,
+  { AVC("timeout"),   OFF(Link.receiveTimeoutInMs),       OPT_INT, 0,
   	"Session timeout in seconds" },
   { AVC("pubUser"),   OFF(Link.pubUser),       OPT_STR, 0,
         "Publisher username" },
@@ -909,6 +908,17 @@ RTMP_Connect0(RTMP *r, struct sockaddr * service)
   r->m_sb.sb_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
   if (r->m_sb.sb_socket != -1)
     {
+      int err;
+      struct timeval send_timeout;
+
+      send_timeout.tv_sec = r->Link.sendTimeoutInMs / 1000;
+      send_timeout.tv_usec = (r->Link.sendTimeoutInMs % 1000) * 1000;
+      err = setsockopt(r->m_sb.sb_socket, SOL_SOCKET, SO_SNDTIMEO, &send_timeout, sizeof(send_timeout));
+      if (err)
+        {
+          RTMP_Log(RTMP_LOGERROR, "Error %d setting SO_SNDTIMEO", errno);
+        }
+
       if (connect(r->m_sb.sb_socket, service, sizeof(struct sockaddr)) < 0)
 	{
 	  int err = GetSockError();
@@ -938,12 +948,15 @@ RTMP_Connect0(RTMP *r, struct sockaddr * service)
 
   /* set timeout */
   {
-    SET_RCVTIMEO(tv, r->Link.timeout);
+    struct timeval tv;
+
+    tv.tv_sec = r->Link.receiveTimeoutInMs / 1000;
+    tv.tv_usec = (r->Link.receiveTimeoutInMs % 1000) * 1000;
     if (setsockopt
         (r->m_sb.sb_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv, sizeof(tv)))
       {
-        RTMP_Log(RTMP_LOGERROR, "%s, Setting socket timeout to %ds failed!",
-	    __FUNCTION__, r->Link.timeout);
+        RTMP_Log(RTMP_LOGERROR, "%s, Setting socket timeout to %dms failed!",
+	    __FUNCTION__, r->Link.receiveTimeoutInMs);
       }
   }
 

rtmp-client/src/main/cpp/librtmp/rtmp.h

@@ -176,7 +176,8 @@ extern "C"
     int swfAge;
 
     int protocol;
-    int timeout;		/* connection timeout in seconds */
+    int receiveTimeoutInMs;
+    int sendTimeoutInMs;
 
 #define RTMP_PUB_NAME   0x0001  /* send login to server */
 #define RTMP_PUB_RESP   0x0002  /* send salted password hash */
@@ -311,7 +312,7 @@ extern "C"
 			AVal *subscribepath,
 			AVal *usherToken,
 			int dStart,
-			int dStop, int bLiveStream, long int timeout);
+			int dStop, int bLiveStream, long int timeoutInMs);
 
   int RTMP_Connect(RTMP *r, RTMPPacket *cp);
   struct sockaddr;

rtmp-client/src/main/java/net/butterflytv/rtmp_client/RtmpClient.java

@@ -13,8 +13,14 @@ public class RtmpClient {
 
 
     private final static int OPEN_SUCCESS = 1;
+    private final static int TIMEOUT_IN_MS = 10000;
     private long rtmpPointer = 0;
 
+    /** Socket send timeout value in milliseconds */
+    private int sendTimeoutInMs = TIMEOUT_IN_MS;
+    /** Socket receive timeout value in seconds */
+    private int receiveTimeoutInMs = TIMEOUT_IN_MS;
+
     public static class RtmpIOException extends IOException {
 
         /**
@@ -48,9 +54,40 @@ public RtmpIOException(int errorCode) {
 
     }
 
+    /**
+     *  Sets the socket's send timeout value
+     * @param sendTimeoutInMs
+     * The send timeout value for the rtmp socket in milliseconds.
+     * Parameter expects a non-zero positive integer and will reset timeout to the default value
+     * (10000 ms) if zero or a negative integer is passed.
+     *  */
+    public void setSendTimeout(int sendTimeoutInMs) {
+        if (sendTimeoutInMs > 0) {
+            this.sendTimeoutInMs = sendTimeoutInMs;
+        } else {
+            this.sendTimeoutInMs = TIMEOUT_IN_MS;
+        }
+    }
+
+    /**
+     *  Sets the socket's receive timeout value
+     * @param receiveTimeoutInMs
+     * The receive timeout value for the rtmp socket in milliseconds.
+     * Parameter expects a non-zero positive integer and will reset timeout to the default value
+     * (10000 ms) if zero or a negative integer is passed.
+     *  */
+    public void setReceiveTimeout(int receiveTimeoutInMs) {
+        if (receiveTimeoutInMs > 0) {
+            this.receiveTimeoutInMs = receiveTimeoutInMs;
+        } else {
+            this.receiveTimeoutInMs = TIMEOUT_IN_MS;
+        }
+    }
+
     public void open(String url, boolean isPublishMode) throws RtmpIOException {
         rtmpPointer = nativeAlloc();
-        int result = nativeOpen(url, isPublishMode, rtmpPointer);
+        int result = nativeOpen(url, isPublishMode, rtmpPointer, sendTimeoutInMs,
+            receiveTimeoutInMs);
         if (result != OPEN_SUCCESS) {
             rtmpPointer = 0;
             throw new RtmpIOException(result);
@@ -72,7 +109,8 @@ public void open(String url, boolean isPublishMode) throws RtmpIOException {
      *
      * returns {@link #OPEN_SUCCESS} if it is successful, throws RtmpIOException if it is failed
      */
-    private native int nativeOpen(String url, boolean isPublishMode, long rtmpPointer);
+    private native int nativeOpen(String url, boolean isPublishMode, long rtmpPointer,
+        int sendTimeoutInMs, int receiveTimeoutInMs);
 
     /**
      * read data from rtmp connection