Merge branch 'develop' into add-event-metadata-support

Author: Yuqi Huang

jni/com/amazonaws/kinesis/video/producer/jni/KinesisVideoClientWrapper.cpp

@@ -33,33 +33,30 @@ KinesisVideoClientWrapper::KinesisVideoClientWrapper(JNIEnv* env,
     if (env->GetJavaVM(&mJVMContext.jvm) != 0 || mJVMContext.jvm == NULL) {
         CHECK_EXT(FALSE, "Couldn't retrieve the JavaVM reference.");
     }
-    clientRegistryInfo = ClientRegistry::getInstance().addClient(this); // Note: 'this' cannot be null
-    mJVMContext.clientId = clientRegistryInfo.second;
 
     // Set the callbacks
     if (!setCallbacks(env, thiz, isFirstClient)) {
         throwNativeException(env, EXCEPTION_NAME, "Failed to set the callbacks.", STATUS_INVALID_ARG);
-        ClientRegistry::getInstance().removeClient(this);
         return;
     }
 
     // Extract the DeviceInfo structure
     MEMSET(&mDeviceInfo, 0x00, SIZEOF(DeviceInfo));
     if (!setDeviceInfo(env, deviceInfo, &mDeviceInfo)) {
         throwNativeException(env, EXCEPTION_NAME, "Failed to set the DeviceInfo structure.", STATUS_INVALID_ARG);
-        ClientRegistry::getInstance().removeClient(this);
         return;
     }
 
     // Creating the client object might return an error as well so freeing potentially allocated tags right after the call.
     retStatus = createKinesisVideoClient(&mDeviceInfo, &mClientCallbacks, &mClientHandle);
     if (STATUS_SUCCEEDED(retStatus) && isFirstClient) {
         firstClientCreated = true;
+        clientRegistryInfo = ClientRegistry::getInstance().addClient(this); // Note: 'this' cannot be null
+        mJVMContext.clientId = clientRegistryInfo.second;
     }
     releaseTags(mDeviceInfo.tags);
     if (STATUS_FAILED(retStatus)) {
         throwNativeException(env, EXCEPTION_NAME, "Failed to create Kinesis Video client.", retStatus);
-        ClientRegistry::getInstance().removeClient(this);
         return;
     }
 

src/main/java/com/amazonaws/kinesisvideo/client/PutMediaClient.java

@@ -105,6 +105,8 @@ private void sign(final ParallelSimpleHttpClient client) {
 
     private Consumer<OutputStream> sendChunkEncodedMvkStream(final int fragmentThrottle) {
         return new Consumer<OutputStream>() {
+
+            // Note: The thread name should already have the streamName + uploadHandle
             @Override
             public void accept(final OutputStream rawOutputStream) {
                 FileOutputStream outputFileStream = null;
@@ -136,10 +138,20 @@ public void accept(final OutputStream rawOutputStream) {
                     rawOutputStream.flush();
                     log.debug("Data sent. counter: {}", counter);
                 } catch (final Exception e) {
-                    log.debug("Exception while sending data.", e);
-                    throw new RuntimeException("Exception while sending encoded chunk in MKV stream ! ", e);
+                    log.debug(mBuilder.mStreamName + ": Exception while sending data.", e);
+                    throw new RuntimeException(mBuilder.mStreamName + ": Exception while sending encoded chunk in MKV stream ! ", e);
                 } finally {
                     tryCloseOutputFileStream(outputFileStream);
+
+                    if (mBuilder.mMkvStream != null) {
+                        try {
+                            log.trace("Closing mkv stream");
+                            mBuilder.mMkvStream.close();
+                            log.trace("Closed the mkv stream");
+                        } catch (final IOException e) {
+                            log.error("Failed to close the MKV stream", e);
+                        }
+                    }
                 }
             }
         };

src/main/java/com/amazonaws/kinesisvideo/http/ParallelSimpleHttpClient.java

@@ -3,12 +3,15 @@
 import com.amazonaws.kinesisvideo.client.IPVersionFilter;
 import com.amazonaws.kinesisvideo.client.KinesisVideoClientConfigurationDefaults;
 import com.amazonaws.kinesisvideo.common.function.Consumer;
+import com.amazonaws.kinesisvideo.common.preconditions.Preconditions;
 import com.amazonaws.kinesisvideo.socket.SocketFactory;
 import com.amazonaws.kinesisvideo.util.LoggedExitRunnable;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.InputStream;
@@ -18,15 +21,21 @@
 import java.net.Socket;
 import java.net.URI;
 import java.nio.charset.Charset;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import static com.amazonaws.kinesisvideo.common.preconditions.Preconditions.checkNotNull;
 
 public final class ParallelSimpleHttpClient implements HttpClient {
+
+    private static final int AWAIT_THREAD_TERMINATE_SECS = 3;
+
     private static final String SPACE = " ";
     private static final String CLRF = "\r\n";
     private static final String HTTP_1_1 = "HTTP/1.1";
@@ -51,6 +60,34 @@ public void accept(final Exception object) {
     private OutputStream mOutputStream;
     private ExecutorService payloadSender;
     private ExecutorService responseReceiver;
+    private final List<ExitResult> exitHistory = new ArrayList<>();
+
+    private enum Caller {
+        SENDER,
+        RECEIVER,
+        CLOSE
+    }
+
+    private static class ExitResult {
+        @Nonnull
+        private Caller caller;
+
+        @Nullable
+        private Exception exception;
+
+        ExitResult(@Nonnull final Caller caller, @Nullable final Exception exception) {
+            this.caller = caller;
+            this.exception = exception;
+        }
+
+        @Override
+        public String toString() {
+            return "ExitResult{" +
+                    "caller=" + caller +
+                    ", exception=" + exception +
+                    '}';
+        }
+    }
 
     private ParallelSimpleHttpClient(final Builder builder) {
         mBuilder = builder;
@@ -157,10 +194,7 @@ public void execute() {
                         log.error("[{}] Exception thrown on sending thread", mBuilder.mStreamName, e);
                         storedException = e;
                     } finally {
-                        //Only call completion if there is an exception, otherwise sender will call completion
-                        if (storedException != null) {
-                            mBuilder.mCompletion.accept(storedException);
-                        }
+                        notifyCompletionCallback(new ExitResult(Caller.SENDER, storedException));
                         payloadSender.shutdownNow();
                     }
                 }
@@ -184,7 +218,7 @@ public void execute() {
                         log.error("[{}] Exception thrown on receiving thread", mBuilder.mStreamName, e);
                         storedException = e;
                     } finally {
-                        mBuilder.mCompletion.accept(storedException);
+                        notifyCompletionCallback(new ExitResult(Caller.RECEIVER, storedException));
                         responseReceiver.shutdownNow();
                         closeSocket();
                     }
@@ -245,7 +279,83 @@ public void close() throws IOException {
         payloadSender.shutdownNow();
         responseReceiver.shutdownNow();
         closeSocket();
-        mBuilder.mCompletion.accept(null);
+
+        awaitTryShutdownThreads();
+
+        notifyCompletionCallback(new ExitResult(Caller.CLOSE, null));
+    }
+
+    // This is used to synchronize the 3 threads which call the completion callback:
+    // - Sender thread
+    // - Receiving ACKs thread
+    // - Thread calling close()
+    // If close() is called, it will immediately invoke the completion callback with success.
+    // Otherwise, it will wait for both sender and receiver threads to exit before notifying.
+    // If applicable, the thread that threw the exception first's result will be propagated.
+    private void notifyCompletionCallback(@Nonnull final ExitResult exitResult) {
+        // Note: the thread name should already have the stream name + connection handle # in it
+        log.debug("Received: {}", exitResult);
+
+        if (mBuilder.mCompletion != null) {
+
+            if (exitResult.caller == Caller.CLOSE) {
+                mBuilder.mCompletion.accept(null);
+                return;
+            }
+
+            Exception exceptionToNotify = null;
+            boolean notify = false;
+            synchronized (this.exitHistory) {
+                this.exitHistory.add(exitResult);
+
+                if (this.exitHistory.size() == 2 &&
+                        ((this.exitHistory.get(0).caller == Caller.SENDER && this.exitHistory.get(1).caller == Caller.RECEIVER) ||
+                        (this.exitHistory.get(0).caller == Caller.RECEIVER && this.exitHistory.get(1).caller == Caller.SENDER))
+                ) {
+                    // Check if either one of them exited with an exception
+                    // If so, propagate it. If both of them terminated normally, notify with null
+                    notify = true;
+
+                    // prioritize the exception that came first
+                    exceptionToNotify = this.exitHistory.get(0).exception;
+                    if (exceptionToNotify == null) {
+                        exceptionToNotify = this.exitHistory.get(1).exception;
+                    }
+                } else {
+                    log.debug("Not notifying this time, caller history: {}", this.exitHistory);
+                }
+            }
+
+            if (notify) {
+                log.debug("[{}] notifying completion callback with {}", mBuilder.mStreamName, exceptionToNotify);
+                mBuilder.mCompletion.accept(exceptionToNotify);
+            }
+        }
+    }
+
+    // Wait for the threads to terminate
+    // If the threads are not alive, returns immediately
+    // Expecting these to be near instantaneous
+    private void awaitTryShutdownThreads() {
+        awaitTermination(this.payloadSender, "payload sender", AWAIT_THREAD_TERMINATE_SECS);
+        awaitTermination(this.responseReceiver, "response receiver", AWAIT_THREAD_TERMINATE_SECS);
+    }
+
+    @SuppressWarnings("ConstantConditions")
+    private void awaitTermination(@Nonnull final ExecutorService executor, @Nonnull final String id,
+                                  final int threadTerminateTimeoutSeconds) {
+        Preconditions.checkArgument(executor != null, "Executor cannot be null");
+        Preconditions.checkArgument(id != null, "ID cannot be null");
+        Preconditions.checkArgument(threadTerminateTimeoutSeconds >= 0, "ThreadTerminateTimeoutSeconds must be positive");
+
+        try {
+            if (!executor.awaitTermination(AWAIT_THREAD_TERMINATE_SECS, TimeUnit.SECONDS)) {
+                log.error("{}: {} couldn't shutdown within {} seconds", mBuilder.mStreamName, id, AWAIT_THREAD_TERMINATE_SECS);
+            }
+        } catch (final InterruptedException e) {
+            log.error("{}: Interrupted while waiting for {} shutdown", mBuilder.mStreamName, id, e);
+            Thread.currentThread().interrupt();
+        }
     }
 
 

src/main/java/com/amazonaws/kinesisvideo/internal/producer/jni/NativeKinesisVideoProducerStream.java

@@ -25,6 +25,7 @@
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Implementation of {@link KinesisVideoProducerStream}
@@ -38,7 +39,7 @@ private class NativeDataInputStream extends InputStream {
         /**
          * Whether the stream has been closed
          */
-        private volatile boolean mStreamClosed = false;
+        private AtomicBoolean mStreamClosed = new AtomicBoolean(false);
 
         // Set the notification values
         private final Object mMonitor = new Object();
@@ -63,7 +64,7 @@ public int read(final byte[] b,
                         final int off,
                         final int len)
                 throws IOException {
-            if (mStreamClosed) {
+            if (mStreamClosed.get()) {
                 mLog.warn("Stream {} with uploadHandle {} has been closed", mStreamInfo.getName(), mUploadHandle);
             }
 
@@ -72,9 +73,9 @@ public int read(final byte[] b,
             // is handled by simply spin-locking until the data is available.
             int bytesRead = -1;
 
-            while (!mStreamClosed) {
+            while (!mStreamClosed.get()) {
                 synchronized (mMonitor) {
-                    while (!mDataAvailable && !mStreamClosed) {
+                    while (!mDataAvailable && !mStreamClosed.get()) {
                         try {
                             mLog.debug("no data for stream {} with uploadHandle {}, waiting", mStreamInfo.getName(),
                                     mUploadHandle);
@@ -87,7 +88,7 @@ public int read(final byte[] b,
 
                     // Clear the availability indicator for now
                     mDataAvailable = false;
-                    if (mStreamClosed) {
+                    if (mStreamClosed.get()) {
                         // Indicate the EOS
                         bytesRead = -1;
                         mLog.debug("Being notified to close stream {} with uploadHandle {}",
@@ -108,7 +109,7 @@ public int read(final byte[] b,
                                 mStreamInfo.getName(), mUploadHandle);
 
                         // Set the flag so the stream is not valid any longer
-                        mStreamClosed = true;
+                        mStreamClosed.set(true);
 
                         if (0 == bytesRead) {
                             // Indicate the EOS
@@ -154,11 +155,11 @@ public int read(final byte[] b)
         public void close()
                 throws IOException
         {
-            // Set the stream to stopped state
-            mStreamClosed = true;
-
-            // Notify the awaiting thread
-            notifyReaderThread(0, 0);
+            // Set the stream to stopped state only once
+            if (mStreamClosed.compareAndSet(false, true)) {
+                // Notify the awaiting thread
+                notifyReaderThread(0, 0);
+            }
         }
 
         protected void notifyReaderThread(final long duration, final long availableSize) {
@@ -178,7 +179,7 @@ protected void endOfReaderThread() {
             // Unblock the awaiting reading code block
             synchronized (mMonitor) {
                 mDataAvailable = true;
-                mStreamClosed = true;
+                mStreamClosed.set(true);
                 mMonitor.notify();
             }
         }

src/main/java/com/amazonaws/kinesisvideo/internal/service/AckConsumer.java

@@ -25,6 +25,7 @@ class AckConsumer implements Consumer<InputStream> {
     private final Logger log;
     private final long uploadHandle;
     private volatile boolean closed = false;
+    private volatile boolean seenErrorAck = false;
 
     public AckConsumer(final long uploadHandle,
                        @Nonnull final KinesisVideoProducerStream stream,
@@ -72,6 +73,11 @@ private void processAckInputStream() {
                     closed = true;
                 } else if (bytesRead != 0) {
                     log.debug("Received ACK bits: {}", bytesString);
+
+                    if (bytesString.contains("ERROR")) {
+                        seenErrorAck = true;
+                    }
+
                     try {
                         stream.parseFragmentAck(uploadHandle, bytesString);
                     } catch (final ProducerException e) {
@@ -90,6 +96,10 @@ private void processAckInputStream() {
         }
     }
 
+    public boolean seenErrorAck() {
+        return this.seenErrorAck;
+    }
+
     public void close() throws ProducerException {
         // Trigger stopping
         closed = true;