-
Notifications
You must be signed in to change notification settings - Fork 51
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Add the capability and example to pipe the output of GetMedia calls to GStreamer. * Add release notes for 1.0.5
- Loading branch information
Showing
23 changed files
with
945 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
124 changes: 124 additions & 0 deletions
124
src/main/java/com/amazonaws/kinesisvideo/parser/examples/ContinuousGetMediaWorker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| /* | ||
| Copyright 2017-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| Licensed under the Apache License, Version 2.0 (the "License"). | ||
| You may not use this file except in compliance with the License. | ||
| A copy of the License is located at | ||
| http://aws.amazon.com/apache2.0/ | ||
| or in the "license" file accompanying this file. | ||
| This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and limitations under the License. | ||
| */ | ||
| package com.amazonaws.kinesisvideo.parser.examples; | ||
|
|
||
| import com.amazonaws.auth.AWSCredentialsProvider; | ||
| import com.amazonaws.client.builder.AwsClientBuilder; | ||
| import com.amazonaws.kinesisvideo.parser.mkv.MkvElementVisitException; | ||
| import com.amazonaws.kinesisvideo.parser.utilities.FragmentMetadata; | ||
| import com.amazonaws.kinesisvideo.parser.utilities.consumer.GetMediaResponseStreamConsumer; | ||
| import com.amazonaws.kinesisvideo.parser.utilities.consumer.GetMediaResponseStreamConsumerFactory; | ||
| import com.amazonaws.regions.Regions; | ||
| import com.amazonaws.services.kinesisvideo.AmazonKinesisVideo; | ||
| import com.amazonaws.services.kinesisvideo.AmazonKinesisVideoMedia; | ||
| import com.amazonaws.services.kinesisvideo.AmazonKinesisVideoMediaClientBuilder; | ||
| import com.amazonaws.services.kinesisvideo.model.APIName; | ||
| import com.amazonaws.services.kinesisvideo.model.GetDataEndpointRequest; | ||
| import com.amazonaws.services.kinesisvideo.model.GetMediaRequest; | ||
| import com.amazonaws.services.kinesisvideo.model.GetMediaResult; | ||
| import com.amazonaws.services.kinesisvideo.model.StartSelector; | ||
| import com.amazonaws.services.kinesisvideo.model.StartSelectorType; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.commons.lang.Validate; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * Worker used to make a GetMedia call to Kinesis Video and stream in data and parse it and apply a visitor. | ||
| */ | ||
| @Slf4j | ||
| public class ContinuousGetMediaWorker extends KinesisVideoCommon implements Runnable { | ||
| private static final int HTTP_STATUS_OK = 200; | ||
| private final AmazonKinesisVideoMedia videoMedia; | ||
| private final GetMediaResponseStreamConsumerFactory consumerFactory; | ||
| private final StartSelector startSelector; | ||
| private Optional<String> fragmentNumberToStartAfter = Optional.empty(); | ||
| private boolean shouldStop = false; | ||
|
|
||
| private ContinuousGetMediaWorker(Regions region, | ||
| AWSCredentialsProvider credentialsProvider, | ||
| String streamName, | ||
| StartSelector startSelector, | ||
| String endPoint, | ||
| GetMediaResponseStreamConsumerFactory consumerFactory) { | ||
| super(region, credentialsProvider, streamName); | ||
|
|
||
| AmazonKinesisVideoMediaClientBuilder builder = AmazonKinesisVideoMediaClientBuilder.standard() | ||
| .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endPoint, region.getName())) | ||
| .withCredentials(getCredentialsProvider()); | ||
|
|
||
| this.videoMedia = builder.build(); | ||
| this.consumerFactory = consumerFactory; | ||
| this.startSelector = startSelector; | ||
| } | ||
|
|
||
| public static ContinuousGetMediaWorker create(Regions region, | ||
| AWSCredentialsProvider credentialsProvider, | ||
| String streamName, | ||
| StartSelector startSelector, | ||
| AmazonKinesisVideo amazonKinesisVideo, | ||
| GetMediaResponseStreamConsumerFactory consumer) { | ||
| String endPoint = amazonKinesisVideo.getDataEndpoint(new GetDataEndpointRequest().withAPIName(APIName.GET_MEDIA) | ||
| .withStreamName(streamName)).getDataEndpoint(); | ||
|
|
||
| return new ContinuousGetMediaWorker(region, credentialsProvider, streamName, startSelector, endPoint, consumer); | ||
| } | ||
|
|
||
| public void stop() { | ||
| shouldStop = true; | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| log.info("Start ContinuousGetMedia worker for stream {}", streamName); | ||
| while (!shouldStop) { | ||
| try { | ||
|
|
||
| StartSelector selectorToUse = fragmentNumberToStartAfter.map(fn -> new StartSelector().withStartSelectorType(StartSelectorType.FRAGMENT_NUMBER) | ||
| .withAfterFragmentNumber(fn)).orElse(startSelector); | ||
|
|
||
| GetMediaResult result = videoMedia.getMedia(new GetMediaRequest().withStreamName(streamName).withStartSelector(selectorToUse)); | ||
| log.info("Start processing GetMedia called for stream {} response {} requestId {}", | ||
| streamName, | ||
| result.getSdkHttpMetadata().getHttpStatusCode(), | ||
| result.getSdkResponseMetadata().getRequestId()); | ||
|
|
||
| if (result.getSdkHttpMetadata().getHttpStatusCode() == HTTP_STATUS_OK) { | ||
| GetMediaResponseStreamConsumer consumer = consumerFactory.createConsumer(); | ||
| consumer.process( | ||
| result.getPayload(), this::updateFragmentNumberToStartAfter); | ||
| } else { | ||
| Thread.sleep(200); | ||
| } | ||
| } catch (IOException | MkvElementVisitException e) { | ||
| log.error("Failure in ContinuousGetMedia worker for stream {} {}", streamName, e); | ||
| } catch (InterruptedException ie) { | ||
| Thread.currentThread().interrupt(); | ||
| throw new RuntimeException(ie); | ||
| } catch (Throwable t) { | ||
| log.error("WHAT",t); | ||
| } finally { | ||
| log.info("Exit processing GetMedia called for stream {}", streamName); | ||
| } | ||
| } | ||
| log.info("Exit ContinuousGetMedia worker for stream {}", streamName); | ||
| } | ||
|
|
||
| private void updateFragmentNumberToStartAfter(FragmentMetadata f) { | ||
| Validate.isTrue(!fragmentNumberToStartAfter.isPresent() | ||
| || f.getFragmentNumberString().compareTo(fragmentNumberToStartAfter.get()) > 0); | ||
| fragmentNumberToStartAfter = Optional.of(f.getFragmentNumberString()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
133 changes: 133 additions & 0 deletions
133
...in/java/com/amazonaws/kinesisvideo/parser/examples/KinesisVideoGStreamerPiperExample.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| /* | ||
| Copyright 2017-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| Licensed under the Apache License, Version 2.0 (the "License"). | ||
| You may not use this file except in compliance with the License. | ||
| A copy of the License is located at | ||
| http://aws.amazon.com/apache2.0/ | ||
| or in the "license" file accompanying this file. | ||
| This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and limitations under the License. | ||
| */ | ||
| package com.amazonaws.kinesisvideo.parser.examples; | ||
|
|
||
| import com.amazonaws.auth.AWSCredentialsProvider; | ||
| import com.amazonaws.kinesisvideo.parser.utilities.consumer.MergedOutputPiperFactory; | ||
| import com.amazonaws.regions.Regions; | ||
| import com.amazonaws.services.kinesisvideo.AmazonKinesisVideo; | ||
| import com.amazonaws.services.kinesisvideo.AmazonKinesisVideoClientBuilder; | ||
| import com.amazonaws.services.kinesisvideo.model.StartSelector; | ||
| import com.amazonaws.services.kinesisvideo.model.StartSelectorType; | ||
| import lombok.Builder; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.commons.lang3.StringUtils; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * Example for continuously piping the output of GetMedia calls from a Kinesis Video stream to GStreamer. | ||
| */ | ||
| @Slf4j | ||
| public class KinesisVideoGStreamerPiperExample extends KinesisVideoCommon { | ||
| private static final String DEFAULT_PATH_TO_GSTREAMER = "/usr/bin/gst-launch-1.0"; | ||
| private static final String [] FDSRC_ARGS = new String[] { | ||
| "-v", | ||
| "fdsrc", | ||
| "!" | ||
| }; | ||
|
|
||
| private final AmazonKinesisVideo amazonKinesisVideo; | ||
| private final InputStream inputStream; | ||
| private final ExecutorService executorService; | ||
| private PutMediaWorker putMediaWorker; | ||
| private final StreamOps streamOps; | ||
| //The arguments to construct the gstreamer pipeline. | ||
| //The merged output of GetMedia will be piped to the gstreamer pipeline created using these arguments. | ||
| private final List<String> gStreamerPipelineArguments; | ||
|
|
||
|
|
||
| @Builder | ||
| private KinesisVideoGStreamerPiperExample(Regions region, | ||
| String streamName, | ||
| AWSCredentialsProvider credentialsProvider, | ||
| InputStream inputVideoStream, | ||
| String gStreamerPipelineArgument) { | ||
| super(region, credentialsProvider, streamName); | ||
| final AmazonKinesisVideoClientBuilder builder = AmazonKinesisVideoClientBuilder.standard(); | ||
| configureClient(builder); | ||
| this.amazonKinesisVideo = builder.build(); | ||
| this.inputStream = inputVideoStream; | ||
| this.streamOps = new StreamOps(region, streamName, credentialsProvider); | ||
| this.executorService = Executors.newFixedThreadPool(2); | ||
| this.gStreamerPipelineArguments = new ArrayList<>(); | ||
| addGStreamerPipelineArguments(gStreamerPipelineArgument); | ||
| } | ||
|
|
||
| private void addGStreamerPipelineArguments(String gStreamerPipeLineArgument) { | ||
| this.gStreamerPipelineArguments.add(pathToExecutable("PATH_TO_GSTREAMER", DEFAULT_PATH_TO_GSTREAMER)); | ||
| addToPipelineArguments(FDSRC_ARGS); | ||
| addToPipelineArguments(gStreamerPipeLineArgument.split("\\s+")); | ||
| } | ||
|
|
||
| private String pathToExecutable(String environmentVariable, String defaultPath) { | ||
| final String environmentVariableValue = System.getenv(environmentVariable); | ||
| return StringUtils.isEmpty(environmentVariableValue) ? defaultPath : environmentVariableValue; | ||
| } | ||
|
|
||
| private void addToPipelineArguments(String []pipelineArguments) { | ||
| for (String pipelineArgument : pipelineArguments) { | ||
| this.gStreamerPipelineArguments.add(pipelineArgument); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * This method executes the example. | ||
| * | ||
| * @throws InterruptedException the thread is interrupted while waiting for the stream to enter the correct state. | ||
| * @throws IOException fails to read video from the input stream or write to the output stream. | ||
| */ | ||
| public void execute () throws InterruptedException, IOException { | ||
| //Create the Kinesis Video stream, deleting and recreating if necessary. | ||
| streamOps.recreateStreamIfNecessary(); | ||
|
|
||
| ContinuousGetMediaWorker getWorker = ContinuousGetMediaWorker.create(getRegion(), | ||
| getCredentialsProvider(), | ||
| getStreamName(), | ||
| new StartSelector().withStartSelectorType(StartSelectorType.EARLIEST), | ||
| amazonKinesisVideo, | ||
| new MergedOutputPiperFactory(Optional.empty(), | ||
| true, | ||
| gStreamerPipelineArguments)); | ||
|
|
||
| executorService.submit(getWorker); | ||
|
|
||
| //Start a PutMedia worker to write data to a Kinesis Video Stream. | ||
| putMediaWorker = PutMediaWorker.create(getRegion(), | ||
| getCredentialsProvider(), | ||
| getStreamName(), | ||
| inputStream, | ||
| amazonKinesisVideo); | ||
| executorService.submit(putMediaWorker); | ||
|
|
||
| Thread.sleep(3000); | ||
| getWorker.stop(); | ||
|
|
||
| executorService.shutdown(); | ||
| executorService.awaitTermination(120, TimeUnit.SECONDS); | ||
| if (!executorService.isTerminated()) { | ||
| log.warn("Shutting down executor service by force"); | ||
| executorService.shutdownNow(); | ||
| } else { | ||
| log.info("Executor service is shutdown"); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.