JNI namespace change

Improve kvssink plugin retry behavior
Unify GetTimeFunc in C++ producer for override
Support for custom user agent string

Author: chehefen

README.md

@@ -44,6 +44,12 @@ This library is licensed under the Amazon Software License.
 
 ----
 ### Release notes
+#### Release 1.6.0 (3rd Dec 2018)
+* JNI namespace change
+* Improve kvssink plugin retry behavior
+* Unify GetTimeFunc in C++ producer for override
+* Support for custom user agent string
+
 #### Release 1.5.1 (28th Sep 2018)
 * Fragment Metadata Support for Producer SDK
 * Add fallback frame rate to avoid negotiation issue on certain machines like macbook air

kinesis-video-gst-demo/kinesis_video_gstreamer_sample_app.cpp

@@ -43,7 +43,7 @@ LOGGER_TAG("com.amazonaws.kinesis.video.gstreamer");
 #define DEFAULT_RECALCULATE_METRICS TRUE
 #define DEFAULT_STREAM_FRAMERATE 25
 #define DEFAULT_AVG_BANDWIDTH_BPS (4 * 1024 * 1024)
-#define DEFAULT_BUFFER_DURATION_SECONDS 180
+#define DEFAULT_BUFFER_DURATION_SECONDS 120
 #define DEFAULT_REPLAY_DURATION_SECONDS 40
 #define DEFAULT_CONNECTION_STALENESS_SECONDS 60
 #define DEFAULT_CODEC_ID "V_MPEG4/ISO/AVC"
@@ -111,7 +111,7 @@ class SampleCredentialProvider : public StaticCredentialProvider {
 
         // Update only the expiration
         auto now_time = std::chrono::duration_cast<std::chrono::seconds>(
-                std::chrono::system_clock::now().time_since_epoch());
+                systemCurrentTime().time_since_epoch());
         auto expiration_seconds = now_time + ROTATION_PERIOD;
         credentials.setExpiration(std::chrono::seconds(expiration_seconds.count()));
         LOG_INFO("New credentials expiration is " << credentials.getExpiration().count());
@@ -122,8 +122,8 @@ class SampleDeviceInfoProvider : public DefaultDeviceInfoProvider {
 public:
     device_info_t getDeviceInfo() override {
         auto device_info = DefaultDeviceInfoProvider::getDeviceInfo();
-        // Set the storage size to 256mb
-        device_info.storageInfo.storageSize = 512 * 1024 * 1024;
+        // Set the storage size to 128mb
+        device_info.storageInfo.storageSize = 128 * 1024 * 1024;
         return device_info;
     }
 };

kinesis-video-gst-demo/kinesis_video_gstreamer_sample_multistream_app.cpp

@@ -47,14 +47,14 @@ LOGGER_TAG("com.amazonaws.kinesis.video.gstreamer");
 #define DEFAULT_RECALCULATE_METRICS TRUE
 #define DEFAULT_STREAM_FRAMERATE 25
 #define DEFAULT_AVG_BANDWIDTH_BPS (4 * 1024 * 1024)
-#define DEFAULT_BUFFER_DURATION_SECONDS 180
+#define DEFAULT_BUFFER_DURATION_SECONDS 120
 #define DEFAULT_REPLAY_DURATION_SECONDS 40
 #define DEFAULT_CONNECTION_STALENESS_SECONDS 60
 #define DEFAULT_CODEC_ID "V_MPEG4/ISO/AVC"
 #define DEFAULT_TRACKNAME "kinesis_video"
 #define APP_SINK_BASE_NAME "appsink"
 #define DEFAULT_BUFFER_SIZE (1 * 1024 * 1024)
-#define DEFAULT_STORAGE_SIZE (512 * 1024 * 1024)
+#define DEFAULT_STORAGE_SIZE (128 * 1024 * 1024)
 #define DEFAULT_ROTATION_TIME_SECONDS 2400
 
 namespace com {
@@ -122,7 +122,7 @@ namespace com {
 
                         // Update only the expiration
                         auto now_time = std::chrono::duration_cast<std::chrono::seconds>(
-                                std::chrono::system_clock::now().time_since_epoch());
+                                systemCurrentTime().time_since_epoch());
                         auto expiration_seconds = now_time + ROTATION_PERIOD;
                         credentials.setExpiration(std::chrono::seconds(expiration_seconds.count()));
                         LOG_INFO("New credentials expiration is " << credentials.getExpiration().count());
@@ -133,7 +133,7 @@ namespace com {
                 public:
                     device_info_t getDeviceInfo() override {
                         auto device_info = DefaultDeviceInfoProvider::getDeviceInfo();
-                        // Set the storage size to 512MB
+                        // Set the storage size to 64MB
                         device_info.storageInfo.storageSize = DEFAULT_STORAGE_SIZE;
                         return device_info;
                     }

kinesis-video-gst-demo/kinesis_video_gstreamer_sample_rtsp_app.cpp

@@ -42,7 +42,7 @@ LOGGER_TAG("com.amazonaws.kinesis.video.gstreamer");
 #define DEFAULT_RECALCULATE_METRICS TRUE
 #define DEFAULT_STREAM_FRAMERATE 25
 #define DEFAULT_AVG_BANDWIDTH_BPS (4 * 1024 * 1024)
-#define DEFAULT_BUFFER_DURATION_SECONDS 180
+#define DEFAULT_BUFFER_DURATION_SECONDS 120
 #define DEFAULT_REPLAY_DURATION_SECONDS 40
 #define DEFAULT_CONNECTION_STALENESS_SECONDS 60
 #define DEFAULT_CODEC_ID "V_MPEG4/ISO/AVC"
@@ -113,7 +113,7 @@ namespace com {
 
                         // Update only the expiration
                         auto now_time = std::chrono::duration_cast<std::chrono::seconds>(
-                                std::chrono::system_clock::now().time_since_epoch());
+                                systemCurrentTime().time_since_epoch());
                         auto expiration_seconds = now_time + ROTATION_PERIOD;
                         credentials.setExpiration(std::chrono::seconds(expiration_seconds.count()));
                         LOG_INFO("New credentials expiration is " << credentials.getExpiration().count());
@@ -124,8 +124,8 @@ namespace com {
                 public:
                     device_info_t getDeviceInfo() override {
                         auto device_info = DefaultDeviceInfoProvider::getDeviceInfo();
-                        // Set the storage size to 512MB
-                        device_info.storageInfo.storageSize = 512 * 1024 * 1024;
+                        // Set the storage size to 128MB
+                        device_info.storageInfo.storageSize = 128 * 1024 * 1024;
                         return device_info;
                     }
                 };

kinesis-video-gstreamer-plugin/demo-src/kvs_producer_plugin_demo.cpp

@@ -206,7 +206,7 @@ int gstreamer_init(int argc, char *argv[]) {
         g_object_set(G_OBJECT (data.source), "do-timestamp", TRUE, "device", "/dev/video0", NULL);
     }
 
-    g_object_set(G_OBJECT (data.kvsproducer), "stream-name", "plugin-test-stream", "frame-timestamp", KVS_SINK_TIMESTAMP_DEFAULT, "storage-size", 512, NULL);
+    g_object_set(G_OBJECT (data.kvsproducer), "stream-name", "plugin-test-stream", "frame-timestamp", KVS_SINK_TIMESTAMP_DEFAULT, "storage-size", 128, NULL);
 
 //    Iot certificate sample usage.
 //    const char* iot_g_struct = "iot-certificate,endpoint=endpoint,"

kinesis-video-gstreamer-plugin/demo-src/kvs_producer_plugin_rtsp_demo.cpp

@@ -136,7 +136,7 @@ int gstreamer_init(int argc, char *argv[]) {
         return 1;
     }
 
-    g_object_set(G_OBJECT (data.kvsproducer), "stream-name", stream_name, "storage-size", 512, NULL);
+    g_object_set(G_OBJECT (data.kvsproducer), "stream-name", stream_name, "storage-size", 128, NULL);
 
 
     g_object_set(G_OBJECT (data.source),

kinesis-video-gstreamer-plugin/plugin-src/ConnectionStaleStateMachine.cpp

@@ -27,7 +27,7 @@ void ConnectionStaleStateMachine::update_timestamp() {
 
 void ConnectionStaleStateMachine::handleConnectionStale() {
     curr_time = std::chrono::duration_cast<std::chrono::milliseconds>(
-            std::chrono::system_clock::now().time_since_epoch());
+            systemCurrentTime().time_since_epoch());
     LOG_INFO("curr_time: " << curr_time.count() << ", quiet_time: " << quiet_time.count()
                            << ", back_to_normal_time: " << back_to_normal_time.count());
     if (quiet_time < curr_time){

kinesis-video-gstreamer-plugin/plugin-src/KvsSinkIotCertCredentialProvider.h

@@ -24,9 +24,12 @@ KvsSinkStreamCallbackProvider::streamErrorReportHandler(UINT64 custom_data, STRE
                                                            UINT64 errored_timecode, STATUS status_code) {
     LOG_ERROR("Reporting stream error. Errored timecode: " << errored_timecode << " Status: 0x" << std::hex << status_code);
     auto customDataObj = reinterpret_cast<CustomData*>(custom_data);
-    if (status_code == static_cast<UINT32>(STATUS_DESCRIBE_STREAM_CALL_FAILED)) {
+    if (status_code == STATUS_DESCRIBE_STREAM_CALL_FAILED ||
+        status_code == STATUS_GET_STREAMING_ENDPOINT_CALL_FAILED) {
         std::lock_guard<std::mutex> lk(customDataObj->closing_stream_handles_queue_mtx);
         customDataObj->closing_stream_handles_queue.push(stream_handle);
+    } else {
+        customDataObj->stream_status = status_code;
     }
 
     return STATUS_SUCCESS;

kinesis-video-gstreamer-plugin/plugin-src/KvsSinkRotatingCredentialProvider.h

@@ -58,7 +58,7 @@ void StreamLatencyStateMachine::doInfiniteRetry() {
 
 void StreamLatencyStateMachine::handleStreamLatency() {
     curr_time = std::chrono::duration_cast<std::chrono::milliseconds>(
-            std::chrono::system_clock::now().time_since_epoch());
+            systemCurrentTime().time_since_epoch());
     LOG_INFO("curr_time: " << curr_time.count() << ", quiet_time: " << quiet_time.count()
                            << ", back_to_normal_time: " << back_to_normal_time.count());
     if (curr_time > back_to_normal_time) {
@@ -85,4 +85,10 @@ void StreamLatencyStateMachine::handleStreamLatency() {
                 break;
         }
     }
-}
+}
+
+StreamLatencyStateMachine::~StreamLatencyStateMachine() {
+    if (undo_throttle.joinable()){
+        undo_throttle.join();
+    }
+}

kinesis-video-gstreamer-plugin/plugin-src/KvsSinkStreamCallbackProvider.cpp

@@ -27,6 +27,9 @@ class StreamLatencyStateMachine {
     StreamLatencyStateMachine(std::shared_ptr<CustomData> data, STREAM_HANDLE stream_handle): current_state(StreamLatencyHandlingState::NORMAL_STATE),
                                                                               data(data), stream_handle(stream_handle) {}
     void handleStreamLatency();
+
+    ~StreamLatencyStateMachine();
+
 private:
     void toResetConnectionState();
     void toThrottlePipelineState();

kinesis-video-gstreamer-plugin/plugin-src/StreamLatencyStateMachine.cpp

@@ -77,28 +77,4 @@ gboolean parseIotCredentialGstructure(GstStructure *g_struct, std::map<std::stri
 CleanUp:
     return ret;
 }
-
-bool parseTimeStr(std::string time_str, std::chrono::duration<uint64_t> &time_obj){
-    bool res = true;
-    std::tm timeinfo = std::tm();
-
-#if defined(__GNUC__) && (__GNUC__ < 5) && !defined(__APPLE__)
-    res = strptime(time_str.c_str(), "%Y-%m-%dT%H:%M:%SZ", &timeinfo) != NULL ? true : false;
-#else
-    std::istringstream iss(time_str);
-    res = iss >> std::get_time(&timeinfo, "%Y-%m-%dT%H:%M:%SZ") ? true : false;
-#endif
-
-    if (res) {
-        std::time_t tt = std::mktime(&timeinfo);
-        // the expiration string from aws credential is in UTC, but get interpreted as local time when mktime.
-        // Thus minus timezone to convert back to UTC.
-        tt -= timezone_offset;
-        std::chrono::system_clock::time_point tp = std::chrono::system_clock::from_time_t (tt);
-        time_obj = std::chrono::duration_cast<std::chrono::seconds>(tp.time_since_epoch());
-        LOG_DEBUG("Credential expiration epoch: " << time_obj.count() << " seconds");
-    }
-
-    return res;
-}
 }

kinesis-video-gstreamer-plugin/plugin-src/StreamLatencyStateMachine.h

@@ -21,7 +21,6 @@ namespace kvs_sink_util{
     gboolean parseIotCredentialGstructure(GstStructure *g_struct,
                                              std::map<std::string, std::string> &iot_cert_params);
 
-    bool parseTimeStr(std::string time_str, std::chrono::duration<uint64_t> &time_obj);
 }
 
 #endif //__KVS_SINK_UTIL_H__