Update StreamPacketizer in base

2 files changed, 33 insertions, 10 deletions

diff --git a/base/src/main/java/com/smartdevicelink/streaming/StreamPacketizer.java b/base/src/main/java/com/smartdevicelink/streaming/StreamPacketizer.java
index 3f7c528f4..7c2a61090 100644
--- a/base/src/main/java/com/smartdevicelink/streaming/StreamPacketizer.java
+++ b/base/src/main/java/com/smartdevicelink/streaming/StreamPacketizer.java
@@ -38,6 +38,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 
 import com.smartdevicelink.SdlConnection.SdlSession;
+import com.smartdevicelink.managers.CompletionListener;
 import com.smartdevicelink.protocol.ProtocolMessage;
 import com.smartdevicelink.protocol.enums.SessionType;
 import com.smartdevicelink.proxy.interfaces.IAudioStreamListener;
@@ -67,7 +68,7 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
     private Object mPauseLock;
     private boolean mPaused;
     private boolean isServiceProtected = false;
-    private BlockingQueue<ByteBuffer> mOutputQueue;
+	private BlockingQueue<ByteBufferWithListener> mOutputQueue;
 
 	public StreamPacketizer(IStreamListener streamListener, InputStream is, SessionType sType, byte rpcSessionID, SdlSession session) throws IOException {
 		super(streamListener, is, sType, rpcSessionID, session);
@@ -82,7 +83,7 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 			bufferSize = BUFF_READ_SIZE;
 			buffer = new byte[bufferSize];
 		}
-		mOutputQueue = new LinkedBlockingQueue<ByteBuffer>(MAX_QUEUE_SIZE / bufferSize);
+		mOutputQueue = new LinkedBlockingQueue<ByteBufferWithListener>(MAX_QUEUE_SIZE / bufferSize);
 	}
 
 	public void start() throws IOException {
@@ -139,9 +140,13 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 						}
 					}
 				} else { // using sendFrame interface
+					ByteBufferWithListener byteBufferWithListener;
 					ByteBuffer frame;
+					CompletionListener completionListener;
 					try {
-						frame = mOutputQueue.take();
+						byteBufferWithListener = mOutputQueue.take();
+						frame = byteBufferWithListener.byteBuffer;
+						completionListener = byteBufferWithListener.completionListener;
 					} catch (InterruptedException e) {
 						Thread.currentThread().interrupt();
 						break;
@@ -164,6 +169,10 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 
 						frame.position(frame.position() + len);
 					}
+
+					if (!frame.hasRemaining() && completionListener != null){
+						completionListener.onComplete(true);
+					}
 				}
 			}
 		} catch (IOException e) 
@@ -213,7 +222,7 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 	 */
 	@Override
 	public void sendFrame(ByteBuffer data, long presentationTimeUs) {
-		sendByteBufferData(data);
+		sendByteBufferData(data, null);
 	}
 
 	/**
@@ -232,9 +241,15 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 	 *
 	 * @see IAudioStreamListener#sendAudio(ByteBuffer, long)
 	 */
+	@Deprecated
 	@Override
 	public void sendAudio(ByteBuffer data, long presentationTimeUs) {
-		sendByteBufferData(data);
+		sendByteBufferData(data, null);
+	}
+
+	@Override
+	public void sendAudio(ByteBuffer data, long presentationTimeUs, CompletionListener completionListener) {
+		sendByteBufferData(data, completionListener);
 	}
 
 	private void sendArrayData(byte[] data, int offset, int length)
@@ -249,13 +264,13 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 		buffer.flip();
 
 		try {
-			mOutputQueue.put(buffer);
+			mOutputQueue.put(new ByteBufferWithListener(buffer, null));
 		} catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
 		}
 	}
 
-	private void sendByteBufferData(ByteBuffer data) {
+	private void sendByteBufferData(ByteBuffer data, CompletionListener completionListener) {
 		if (data == null || data.remaining() == 0) {
 			return;
 		}
@@ -267,9 +282,19 @@ public class StreamPacketizer extends AbstractPacketizer implements IVideoStream
 		buffer.flip();
 
 		try {
-			mOutputQueue.put(buffer);
+			mOutputQueue.put(new ByteBufferWithListener(buffer, completionListener));
 		} catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
 		}
 	}
+
+
+	private class ByteBufferWithListener{
+		ByteBuffer byteBuffer;
+		CompletionListener completionListener;
+		ByteBufferWithListener (ByteBuffer byteBuffer, CompletionListener completionListener){
+			this.byteBuffer = byteBuffer;
+			this.completionListener = completionListener;
+		}
+	}
 }
diff --git a/hello_sdl_java/src/main/java/com/smartdevicelink/java/SdlService.java b/hello_sdl_java/src/main/java/com/smartdevicelink/java/SdlService.java
index 9af923026..9b9c9afd2 100644
--- a/hello_sdl_java/src/main/java/com/smartdevicelink/java/SdlService.java
+++ b/hello_sdl_java/src/main/java/com/smartdevicelink/java/SdlService.java
@@ -55,8 +55,6 @@ import com.smartdevicelink.util.DebugTool;
 
 import java.util.*;
 
-import static javafx.scene.input.KeyCode.R;
-
 public class SdlService {